143 lines
3.7 KiB
Python
143 lines
3.7 KiB
Python
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from dns_sync.adapters import (
|
|
CERTIFICATES,
|
|
PROXY_HOSTS,
|
|
NormalizedObject,
|
|
ObjectReference,
|
|
Observation,
|
|
)
|
|
from dns_sync.persistence import SQLiteStateStore
|
|
|
|
|
|
ENDPOINT = "npm://192.168.1.4:81"
|
|
NOW = datetime(2026, 9, 20, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def local_certificate_reference() -> ObjectReference:
|
|
return ObjectReference(
|
|
relation="certificate",
|
|
object_type=CERTIFICATES,
|
|
local_key="67",
|
|
)
|
|
|
|
|
|
def logical_certificate_reference() -> ObjectReference:
|
|
return ObjectReference(
|
|
relation="certificate",
|
|
object_type=CERTIFICATES,
|
|
logical_key="app.example.com",
|
|
)
|
|
|
|
|
|
def proxy_host(
|
|
*,
|
|
local_key: str | None,
|
|
reference: ObjectReference,
|
|
) -> NormalizedObject:
|
|
return NormalizedObject(
|
|
object_type=PROXY_HOSTS,
|
|
logical_key="app.example.com",
|
|
local_key=local_key,
|
|
payload={
|
|
"domain_names": ["app.example.com"],
|
|
"forward_scheme": "http",
|
|
"forward_host": "192.168.1.20",
|
|
"forward_port": 8080,
|
|
"enabled": True,
|
|
},
|
|
references=(reference,),
|
|
)
|
|
|
|
|
|
class ReferencePersistenceTests(unittest.TestCase):
|
|
def test_observed_local_reference_round_trips(self):
|
|
observed = Observation(
|
|
endpoint_id=ENDPOINT,
|
|
object=proxy_host(
|
|
local_key="12",
|
|
reference=local_certificate_reference(),
|
|
),
|
|
observed_at=NOW,
|
|
)
|
|
|
|
with SQLiteStateStore(":memory:") as store:
|
|
store.replace_observations(
|
|
ENDPOINT,
|
|
PROXY_HOSTS,
|
|
[observed],
|
|
NOW,
|
|
)
|
|
loaded = store.load_observations(PROXY_HOSTS)
|
|
|
|
self.assertEqual(loaded, [observed])
|
|
self.assertEqual(
|
|
loaded[0].object.references[0].local_key,
|
|
"67",
|
|
)
|
|
|
|
def test_desired_logical_reference_round_trips(self):
|
|
desired = proxy_host(
|
|
local_key=None,
|
|
reference=logical_certificate_reference(),
|
|
)
|
|
|
|
with SQLiteStateStore(":memory:") as store:
|
|
store.replace_desired_state(
|
|
PROXY_HOSTS,
|
|
[desired],
|
|
NOW,
|
|
)
|
|
loaded = store.load_desired_state(PROXY_HOSTS)
|
|
|
|
self.assertEqual(loaded[0].object, desired)
|
|
self.assertEqual(
|
|
loaded[0].object.references[0].logical_key,
|
|
"app.example.com",
|
|
)
|
|
|
|
def test_applied_reference_participates_in_verification(self):
|
|
applied = proxy_host(
|
|
local_key="22",
|
|
reference=logical_certificate_reference(),
|
|
)
|
|
different_reference = proxy_host(
|
|
local_key="22",
|
|
reference=ObjectReference(
|
|
relation="certificate",
|
|
object_type=CERTIFICATES,
|
|
logical_key="other.example.com",
|
|
),
|
|
)
|
|
|
|
with SQLiteStateStore(":memory:") as store:
|
|
store.record_applied(ENDPOINT, applied, NOW)
|
|
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"does not match",
|
|
):
|
|
store.record_verification_success(
|
|
ENDPOINT,
|
|
different_reference,
|
|
NOW,
|
|
)
|
|
|
|
store.record_verification_success(
|
|
ENDPOINT,
|
|
applied,
|
|
NOW,
|
|
)
|
|
loaded = store.load_applied_state(
|
|
ENDPOINT,
|
|
PROXY_HOSTS,
|
|
)
|
|
|
|
self.assertEqual(loaded[0].object, applied)
|
|
self.assertTrue(loaded[0].is_verified)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|