126 lines
3.7 KiB
Python
126 lines
3.7 KiB
Python
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",
|
|
"tailscale_hostname": "test-host",
|
|
"tailscale_name": "name",
|
|
"hostname": "test-host",
|
|
"instance": "test-host",
|
|
}
|
|
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) == []
|