276 lines
11 KiB
Python
276 lines
11 KiB
Python
import copy
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from deploy import aliyun_host as cloud
|
|
from deploy import render_asterisk as ast
|
|
|
|
|
|
def config():
|
|
return {
|
|
"region": cloud.REGION,
|
|
"public_ip": cloud.PUBLIC_IP,
|
|
"project_tag": "ai-call",
|
|
"image_id": "m-test",
|
|
"instance_type": "ecs.test",
|
|
"vswitch_id": "vsw-test",
|
|
"security_group_id": "sg-test",
|
|
"key_pair_name": "test-key",
|
|
"spot_price_limit": 0.1,
|
|
}
|
|
|
|
|
|
def instance(id="i-test", tagged=True):
|
|
return {
|
|
"InstanceId": id,
|
|
"Status": "Running",
|
|
"Tags": {
|
|
"Tag": [{"TagKey": "project", "TagValue": "ai-call"}] if tagged else []
|
|
},
|
|
}
|
|
|
|
|
|
class FakeCloud:
|
|
def __init__(self):
|
|
self.address = {
|
|
"IpAddress": cloud.PUBLIC_IP,
|
|
"AllocationId": "eip-test",
|
|
"Status": "Available",
|
|
"InstanceId": "",
|
|
"InstanceType": "EcsInstance",
|
|
}
|
|
self.rows = []
|
|
self.calls = []
|
|
self.fail_bind = False
|
|
self.timeout_create = False
|
|
|
|
def __call__(self, product, action, **params):
|
|
self.calls.append((action, params))
|
|
if action == "DescribeEipAddresses":
|
|
rows = [copy.deepcopy(self.address)] if self.address else []
|
|
return {"TotalCount": len(rows), "EipAddresses": {"EipAddress": rows}}
|
|
if action == "DescribeInstances":
|
|
rows = copy.deepcopy(self.rows)
|
|
if "InstanceIds" in params:
|
|
rows = [r for r in rows if r["InstanceId"] in params["InstanceIds"]]
|
|
return {"TotalCount": len(rows), "Instances": {"Instance": rows}}
|
|
if action == "RunInstances":
|
|
if self.timeout_create:
|
|
raise cloud.CloudError("simulated transport timeout")
|
|
self.rows = [instance("i-created")]
|
|
return {"InstanceIdSets": {"InstanceIdSet": ["i-created"]}}
|
|
if action == "AssociateEipAddress":
|
|
if self.fail_bind:
|
|
raise cloud.CloudError("simulated association failure")
|
|
self.address.update(InstanceId=params["InstanceId"], Status="InUse")
|
|
return {}
|
|
raise AssertionError("Unexpected cloud mutation: " + action)
|
|
|
|
def mutations(self):
|
|
return [name for name, _ in self.calls if not name.startswith("Describe")]
|
|
|
|
|
|
class CloudTests(unittest.TestCase):
|
|
def test_plan_is_read_only(self):
|
|
api = FakeCloud()
|
|
self.assertEqual(cloud.plan(config(), api)["action"], "create_and_bind")
|
|
self.assertEqual(api.mutations(), [])
|
|
|
|
def test_reuse_owned_bound_instance(self):
|
|
api = FakeCloud()
|
|
api.rows = [instance()]
|
|
api.address.update(InstanceId="i-test", Status="InUse")
|
|
with tempfile.TemporaryDirectory() as d:
|
|
result = cloud.apply(config(), api, Path(d) / "state.json")
|
|
self.assertEqual(result["action"], "reuse")
|
|
self.assertEqual(api.mutations(), [])
|
|
|
|
def test_foreign_attachment_is_never_stolen(self):
|
|
api = FakeCloud()
|
|
api.rows = [instance(tagged=False)]
|
|
api.address.update(InstanceId="i-test", Status="InUse")
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.plan(config(), api)
|
|
self.assertEqual(api.mutations(), [])
|
|
|
|
def test_no_fixed_ip_means_no_creation(self):
|
|
api = FakeCloud()
|
|
api.address = {}
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.plan(config(), api)
|
|
self.assertEqual(api.mutations(), [])
|
|
|
|
def test_existing_non_eip_needs_explicit_adoption(self):
|
|
api = FakeCloud()
|
|
api.address = {}
|
|
row = instance(tagged=False)
|
|
row["PublicIpAddress"] = {"IpAddress": [cloud.PUBLIC_IP]}
|
|
api.rows = [row]
|
|
cfg = config()
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.plan(cfg, api)
|
|
cfg["adopt_instance_id"] = "i-test"
|
|
self.assertEqual(cloud.plan(cfg, api)["ip_kind"], "instance_public_ip")
|
|
|
|
def test_duplicate_candidates_stop(self):
|
|
api = FakeCloud()
|
|
api.rows = [instance("i-one"), instance("i-two")]
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.plan(config(), api)
|
|
|
|
def test_create_spot_without_new_public_ip_then_bind(self):
|
|
api = FakeCloud()
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = Path(d) / "state.json"
|
|
result = cloud.apply(config(), api, path, sleep=lambda _: None)
|
|
state = json.loads(path.read_text())
|
|
self.assertEqual(state["instance_id"], "i-created")
|
|
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
|
|
self.assertEqual(result["action"], "ready")
|
|
params = next(params for name, params in api.calls if name == "RunInstances")
|
|
self.assertEqual(params["Amount"], 1)
|
|
self.assertEqual(params["InternetMaxBandwidthOut"], 0)
|
|
self.assertEqual(params["SpotStrategy"], "SpotWithPriceLimit")
|
|
self.assertEqual(params["SpotPriceLimit"], 0.1)
|
|
self.assertEqual(api.mutations(), ["RunInstances", "AssociateEipAddress"])
|
|
|
|
def test_create_timeout_reuses_client_token(self):
|
|
api = FakeCloud()
|
|
api.timeout_create = True
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = Path(d) / "state.json"
|
|
for _ in range(2):
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.apply(config(), api, path, sleep=lambda _: None)
|
|
tokens = [p["ClientToken"] for name, p in api.calls if name == "RunInstances"]
|
|
self.assertEqual(len(tokens), 2)
|
|
self.assertEqual(tokens[0], tokens[1])
|
|
|
|
def test_bind_failure_reuses_created_host(self):
|
|
api = FakeCloud()
|
|
api.fail_bind = True
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = Path(d) / "state.json"
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.apply(config(), api, path, sleep=lambda _: None)
|
|
api.fail_bind = False
|
|
result = cloud.apply(config(), api, path, sleep=lambda _: None)
|
|
self.assertEqual(result["action"], "ready")
|
|
self.assertEqual(api.mutations().count("RunInstances"), 1)
|
|
|
|
def test_stopped_host_and_bad_budget_do_not_create(self):
|
|
api = FakeCloud()
|
|
cfg = config()
|
|
cfg["spot_price_limit"] = None
|
|
with tempfile.TemporaryDirectory() as d, self.assertRaises(cloud.CloudError):
|
|
cloud.apply(cfg, api, Path(d) / "state.json", sleep=lambda _: None)
|
|
self.assertEqual(api.mutations(), [])
|
|
api.rows = [instance()]
|
|
api.rows[0]["Status"] = "Stopped"
|
|
with tempfile.TemporaryDirectory() as d, self.assertRaises(cloud.CloudError):
|
|
cloud.apply(config(), api, Path(d) / "state.json", sleep=lambda _: None)
|
|
self.assertEqual(api.mutations(), [])
|
|
|
|
def test_incomplete_inventory_does_not_create(self):
|
|
api = FakeCloud()
|
|
|
|
def incomplete(product, action, **params):
|
|
if action == "DescribeInstances":
|
|
return {"TotalCount": 2, "Instances": {"Instance": []}}
|
|
return api(product, action, **params)
|
|
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.plan(config(), incomplete)
|
|
self.assertEqual(api.mutations(), [])
|
|
|
|
def test_missing_allocation_id_stops_before_creation(self):
|
|
api = FakeCloud()
|
|
api.address.pop("AllocationId")
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.plan(config(), api)
|
|
self.assertEqual(api.mutations(), [])
|
|
|
|
def test_binding_race_does_not_detach_foreign_host(self):
|
|
api = FakeCloud()
|
|
reads = 0
|
|
|
|
def raced(product, action, **params):
|
|
nonlocal reads
|
|
if action == "DescribeEipAddresses":
|
|
reads += 1
|
|
if reads == 2:
|
|
api.address.update(InstanceId="i-foreign", Status="InUse")
|
|
return api(product, action, **params)
|
|
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = Path(d) / "state.json"
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.apply(config(), raced, path, sleep=lambda _: None)
|
|
self.assertEqual(json.loads(path.read_text())["instance_id"], "i-created")
|
|
self.assertEqual(api.mutations(), ["RunInstances"])
|
|
|
|
def test_region_mismatch(self):
|
|
cfg = config()
|
|
cfg["region"] = "cn-hangzhou"
|
|
with self.assertRaises(cloud.CloudError):
|
|
cloud.plan(cfg, FakeCloud())
|
|
|
|
|
|
class AsteriskTests(unittest.TestCase):
|
|
def cfg(self):
|
|
return {
|
|
"public_ip": cloud.PUBLIC_IP,
|
|
"transport": "udp",
|
|
"local_net": "10.1.0.0/16",
|
|
"primary": {"host": "sip-a.test", "auth_mode": "ip"},
|
|
"backup": {"host": "sip-b.test", "auth_mode": "ip"},
|
|
}
|
|
|
|
def test_private_ari_fixed_nat_and_recording_config(self):
|
|
files = ast.render(self.cfg(), {"ARI_PASSWORD": "x" * 32})
|
|
self.assertEqual(len(files), 5)
|
|
self.assertIn("bindaddr=127.0.0.1", files["http.conf"])
|
|
self.assertIn("external_media_address=123.56.71.98", files["pjsip.conf"])
|
|
self.assertIn("context=deny-inbound", files["pjsip.conf"])
|
|
self.assertIn("strictrtp=yes", files["rtp.conf"])
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = Path(d) / "generated"
|
|
ast.write_config(files, path)
|
|
self.assertEqual((path / "ari.conf").stat().st_mode & 0o777, 0o600)
|
|
with self.assertRaises(ValueError):
|
|
ast.write_config(files, path)
|
|
|
|
def test_reject_injection_public_ari_and_our_ip_as_provider(self):
|
|
for update in (
|
|
{"ari_bind": "0.0.0.0"},
|
|
{"ari_bind": "8.8.8.8"},
|
|
{"local_net": "0.0.0.0/0"},
|
|
): # noqa: S104 — negative fixtures; renderer must reject them.
|
|
cfg = self.cfg()
|
|
cfg.update(update)
|
|
with self.assertRaises(ValueError):
|
|
ast.render(cfg, {"ARI_PASSWORD": "x" * 32})
|
|
for host in ("123.56.71.98", "host\n[evil]", ""):
|
|
cfg = self.cfg()
|
|
cfg["primary"]["host"] = host
|
|
with self.assertRaises(ValueError):
|
|
ast.render(cfg, {"ARI_PASSWORD": "x" * 32})
|
|
|
|
def test_digest_registration(self):
|
|
cfg = self.cfg()
|
|
cfg["primary"].update(
|
|
auth_mode="digest", username="approved-user", register=True
|
|
)
|
|
files = ast.render(
|
|
cfg,
|
|
{"ARI_PASSWORD": "x" * 32, "SIP_PRIMARY_PASSWORD": "vendor-pass-123456"},
|
|
)
|
|
self.assertIn("outbound_auth=provider-primary-auth", files["pjsip.conf"])
|
|
self.assertIn("type=registration", files["pjsip.conf"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|