from tailscalesd.main import group_by_type, matrix_workers_to_sd def test_group_by_type_valid_input(): input_list = [ {"type": "A", "value": 1}, {"type": "B", "value": 2}, {"type": "A", "value": 3}, {"type": "C", "value": 5}, ] expected_output = { "A": [{"type": "A", "value": 1}, {"type": "A", "value": 3}], "B": [{"type": "B", "value": 2}], "C": [{"type": "C", "value": 5}], } assert group_by_type(input_list) == expected_output def test_group_by_type_missing_type_key(): input_list = [ {"value": 1}, {"type": "B", "value": 2}, {"value": 3}, {"type": "C", "value": 5}, ] expected_output = { "B": [{"type": "B", "value": 2}], "C": [{"type": "C", "value": 5}], } assert group_by_type(input_list) == expected_output def test_group_by_type_empty_input(): input_list = [] expected_output = {} assert group_by_type(input_list) == expected_output def test_matrix_workers_to_sd_valid_input(): device = { "addresses": ["10.0.0.1", "192.168.1.1"], "hostname": "test-host", "clientVersion": "1", "id": "id", "name": "name", "os": "linux", "authorized": True, "tags": ["tag:test"], "tailnet": "testnet", } workers = { "type1": [ {"metrics_port": 1234, "name": "worker1"}, {"metrics_port": 1235, "name": "worker2"}, ], "type2": [ {"metrics_port": 1236, "name": "worker3"}, ], } ts_labels = { "__meta_tailscale_device_client_version": "1", "__meta_tailscale_device_hostname": "test-host", "__meta_tailscale_device_authorized": "true", "__meta_tailscale_device_id": "id", "__meta_tailscale_device_name": "name", "__meta_tailscale_device_os": "linux", "__meta_tailscale_tailnet": "testnet", } expected_output = [ { "targets": ["10.0.0.1:1234"], "labels": { # Add expected tailscale_labels output here "__meta_matrix_worker_type": "type1", "__meta_matrix_worker_name": "worker1", } | ts_labels, }, { "targets": ["10.0.0.1:1235"], "labels": { # Add expected tailscale_labels output here "__meta_matrix_worker_type": "type1", "__meta_matrix_worker_name": "worker2", } | ts_labels, }, { "targets": ["10.0.0.1:1236"], "labels": { # Add expected tailscale_labels output here "__meta_matrix_worker_type": "type2", "__meta_matrix_worker_name": "worker3", } | ts_labels, }, ] actual_output = matrix_workers_to_sd("testnet", device, workers) assert actual_output == expected_output def test_matrix_workers_to_sd_empty_workers(): device = { "addresses": ["10.0.0.1", "192.168.1.1"], "hostname": "test-host", } workers = {} assert matrix_workers_to_sd("testnet", device, workers) == [] def test_matrix_workers_to_sd_invalid_port(): device = { "addresses": ["10.0.0.1", "192.168.1.1"], "hostname": "test-host", } workers = { "type1": [ {"metrics_port": None, "name": "worker1"}, ], } # Assuming log.error does not raise any exceptions assert matrix_workers_to_sd("testnet", device, workers) == []