63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
import asyncio
|
|
from datetime import datetime
|
|
|
|
from idea139_opendeployment_quilt_federated import models, registry, coordinator
|
|
from idea139_opendeployment_quilt_federated.adapters.github_actions import GitHubActionsAdapter
|
|
from idea139_opendeployment_quilt_federated.adapters.jenkins import JenkinsAdapter
|
|
|
|
|
|
def test_models_validation():
|
|
lp = models.LocalPlan(id="p1", service="svc", pipeline={"jobs": []})
|
|
assert lp.service == "svc"
|
|
|
|
ss = models.SharedSignals(service="svc", metrics={"failure_rate": 0.1})
|
|
assert "failure_rate" in ss.metrics
|
|
|
|
pd = models.PlanDelta(id="d1", plan_id="p1", delta={"action": "no_change"}, author_id="a")
|
|
assert pd.plan_id == "p1"
|
|
|
|
|
|
def test_registry_and_adapters():
|
|
reg = registry.registry
|
|
# ensure clean state for test
|
|
# it's fine if adapters already registered; we just test registration API
|
|
|
|
ga = GitHubActionsAdapter(repo="org/repo")
|
|
jn = JenkinsAdapter(jenkins_url="http://jenkins.local")
|
|
|
|
# register temporary adapters with unique names
|
|
name1 = "test-ga"
|
|
name2 = "test-jn"
|
|
# unregister if present (be defensive)
|
|
if name1 in reg.list_adapters():
|
|
# skip registration if already present
|
|
pass
|
|
else:
|
|
reg.register_adapter(name1, ga, registry.AdapterContract(name="ga", version="0.1"))
|
|
|
|
if name2 in reg.list_adapters():
|
|
pass
|
|
else:
|
|
reg.register_adapter(name2, jn, registry.AdapterContract(name="jn", version="0.1"))
|
|
|
|
assert name1 in reg.list_adapters()
|
|
assert name2 in reg.list_adapters()
|
|
|
|
|
|
def test_coordinator_aggregate_and_propose():
|
|
coord = coordinator.SimpleCoordinator()
|
|
|
|
async def run():
|
|
await coord.append_signal(models.SharedSignals(service="svc", metrics={"failure_rate": 0.02}))
|
|
await coord.append_signal(models.SharedSignals(service="svc", metrics={"failure_rate": 0.1}))
|
|
agg = await coord.aggregate()
|
|
assert "failure_rate" in agg
|
|
# average of .02 and .1 -> 0.06
|
|
assert 0.059 < agg["failure_rate"] < 0.061
|
|
|
|
pd = await coord.propose_delta(service="svc")
|
|
# since average > 0.05, expect an add_canary action
|
|
assert pd.delta["action"] in ("add_canary", "no_change")
|
|
|
|
asyncio.get_event_loop().run_until_complete(run())
|