AkurAI-EC2 / tests/test_ops_helpers.py

Latest change d8aa29121486f4b02d0016888d94fa66cac36ffe - Checkpoint WIP + rustfmt + AkurAI Build CI/CD pipeline by Ólafur Búi Ólafsson

#!/usr/bin/env python3
from __future__ import annotations

import importlib.util
import json
import pathlib
import subprocess
import sys
import tempfile
import unittest

ROOT = pathlib.Path(__file__).resolve().parents[1]


def load(name: str, path: pathlib.Path):
    spec = importlib.util.spec_from_file_location(name, path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


class DnsHelperTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.dns = load("akurai_ec2_dns", ROOT / "libexec/akurai-ec2-dns.py")

    def test_zone_page_records_are_structured(self) -> None:
        class FixtureClient(self.dns.Client):
            def request(self, path, data=None, referer=None):
                if path == "/domains":
                    return '<a href="/domains/zone/42">olibuijr.com</a>'
                return """<table><tr id="entry_77"><td>bunfork</td><td>A</td>
                    <td>900</td><td>3.94.46.219</td></tr></table>"""

        records = FixtureClient().records("olibuijr.com")
        self.assertEqual(len(records), 1)
        self.assertEqual(records[0].id, "77")
        self.assertEqual(records[0].fqdn, "bunfork.olibuijr.com")
        self.assertEqual(records[0].type, "A")
        self.assertEqual(records[0].ttl, 900)

    def test_delete_requires_yes_before_login(self) -> None:
        result = subprocess.run([
            str(ROOT / "bin/akurai-ec2"), "dns", "delete", "bunfork.olibuijr.com", "--type", "A"
        ], text=True, capture_output=True)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("requires --yes", result.stderr)

    def test_reserved_domain_verifies_absent_without_credentials(self) -> None:
        result = subprocess.run([
            str(ROOT / "bin/akurai-ec2"), "dns", "verify", "akurai.invalid", "--type", "A", "--absent"
        ], text=True, capture_output=True, check=True)
        payload = json.loads(result.stdout)
        self.assertTrue(payload["ok"])
        self.assertEqual(payload["state"], "absent")


class RetirementHelperTests(unittest.TestCase):
    def test_apply_requires_yes_before_remote_access(self) -> None:
        result = subprocess.run([
            str(ROOT / "bin/akurai-ec2"), "retire-app", "example", "--domain", "example.invalid", "--apply"
        ], text=True, capture_output=True)
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("requires --yes", result.stderr)

    def test_copy_install_includes_every_helper(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            target = pathlib.Path(tmp) / "bin/akurai-ec2"
            subprocess.run([str(ROOT / "install.sh"), "--copy"], env={
                "PATH": str(pathlib.Path("/usr/bin")),
                "HOME": tmp,
                "AKURAI_EC2_INSTALL_TARGET": str(target),
            }, check=True, text=True, capture_output=True)
            helper_dir = pathlib.Path(tmp) / "libexec"
            for name in ("akurai-ec2-manifest.py", "akurai-ec2-dns.py", "akurai-ec2-retire-app.py"):
                self.assertTrue((helper_dir / name).is_file(), name)


if __name__ == "__main__":
    unittest.main()