scripts/test-queue-runtime-qa.py

Rama main · réplica de lectura del repositorio original.

#!/usr/bin/env python3
"""Queue QA target guards and failure handling; no VM or network changes."""
import importlib.util
import io
import json
import sqlite3
import subprocess
import tempfile
import uuid
from pathlib import Path
import unittest
from unittest.mock import patch

spec = importlib.util.spec_from_file_location("queue_qa", Path(__file__).with_name("ci-qa-queue.py"))
qa = importlib.util.module_from_spec(spec)
spec.loader.exec_module(qa)


class QueueQaContracts(unittest.TestCase):
    def test_only_explicit_local_http_origins_are_accepted(self):
        for value in ["https://127.0.0.1:5523", "http://remote:5523", "http://user:pass@localhost:5523",
                      "http://localhost", "http://localhost:5523/api", "http://localhost:5523/?token=x"]:
            with self.subTest(value=value), self.assertRaises(ValueError):
                qa.local_origin(value)
        self.assertEqual(qa.local_origin("http://127.0.0.1:15523/"), "http://127.0.0.1:15523")

    def test_subnets_must_be_private_ipv4_with_bounded_capacity(self):
        for value in ["8.8.8.0/24", "127.0.0.0/24", "10.0.0.0/16", "192.168.1.0/30", "fd00::/64"]:
            with self.subTest(value=value), self.assertRaises(ValueError):
                qa.private_subnet(value)
        self.assertEqual(qa.private_subnet("172.28.249.0/24"), "172.28.249.0/24")

    def test_existing_engine_is_refused_before_any_mutation(self):
        with patch.object(qa, "api", return_value={"vmId": "existing"}) as api:
            with self.assertRaisesRegex(RuntimeError, "no existing engine"):
                qa.run(None, {"checks": []})
        api.assert_called_once_with("/api/queue/engine")

    def test_permanent_engine_failure_does_not_wait_twenty_minutes(self):
        with patch.object(qa, "api", return_value={"phase": "failed", "error": "Image import failed"}):
            with self.assertRaisesRegex(qa.EngineFailed, "Image import failed"):
                qa.wait("engine boot", lambda: qa.engine_phase("running"))

    def test_disk_probe_reads_identity_without_returning_private_columns(self):
        with tempfile.TemporaryDirectory() as root:
            database = Path(root) / "state.db"
            vm_id, generation = uuid.uuid4(), uuid.uuid4()
            with sqlite3.connect(database) as db:
                db.execute("CREATE TABLE vms(id TEXT, storage_root TEXT, disk_generation TEXT, purpose TEXT, env TEXT)")
                db.execute("INSERT INTO vms VALUES(?,?,?,?,?)", (str(vm_id), "default", str(generation), "queue", "private-value"))
            disk = database.parent / "vms" / vm_id.hex / "d" / (generation.hex + ".ext4")
            disk.parent.mkdir(parents=True)
            disk.write_bytes(b"persistent-data")
            run = subprocess.run
            with patch.object(qa.subprocess, "run", side_effect=lambda command, **kw: run(command[2:], **kw)):
                identity = qa.disk_identity(database, str(vm_id))
                self.assertEqual(identity, {"generation": str(generation), "bytes": 15})
                self.assertNotIn("private", json.dumps(identity))
                disk.unlink()
                with self.assertRaisesRegex(RuntimeError, "disk identity"):
                    qa.disk_identity(database, str(vm_id))

    def test_http_redirects_are_not_followed(self):
        self.assertIsNone(qa.NoRedirect().redirect_request(None, None, 302, "redirect", {}, "http://remote"))

    def test_http_failure_reports_validation_fields_without_private_values(self):
        body = {"error": {"code": "validation_failed", "message": "private-message",
                          "fields": {"storageRoot": "private-value"}}}
        error = qa.urllib.error.HTTPError(qa.ORIGIN, 400, "bad request", {}, io.BytesIO(json.dumps(body).encode()))
        with patch.object(qa.OPENER, "open", side_effect=error):
            with self.assertRaisesRegex(RuntimeError, "validation_failed; fields: storageRoot") as caught:
                qa.api("/api/queue/engine", "POST", {})
        self.assertNotIn("private", str(caught.exception))
        self.assertTrue(error.closed)

    def test_malformed_http_failure_keeps_the_status_without_response_contents(self):
        error = qa.urllib.error.HTTPError(qa.ORIGIN, 500, "bad response", {}, io.BytesIO(b"private-value"))
        with patch.object(qa.OPENER, "open", side_effect=error):
            with self.assertRaisesRegex(RuntimeError, r"HTTP 500$"):
                qa.api("/api/queue/engine")
        self.assertTrue(error.closed)


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