test_operational_edges.py #1

#!/usr/bin/env python3
"""Offline configuration-generation and cloud-entry contracts; no live APIs."""
from pathlib import Path
import os
import shutil
import subprocess
import tempfile
import unittest

SDP = Path(__file__).resolve().parents[1]
BIN = SDP / "Server/Unix/p4/common/bin"
BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash"))


class OperationalEdges(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="sdp-operational-edges-")
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.env = dict(os.environ, TEST_ROOT=str(self.root))

    def executable(self, path, body):
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text("#!" + BASH + "\n" + body)
        path.chmod(0o755)

    def run_script(self, path, args=(), extra=None):
        return subprocess.run([BASH, path, *args], env=dict(self.env, **(extra or {})),
                              capture_output=True, text=True, timeout=10)

    def broker_fixture(self):
        common = self.root / "common/bin"
        common.mkdir(parents=True)
        (self.root / "database").mkdir()
        (common / "p4_vars").write_text('''P4ROOT="$TEST_ROOT/database"
P4DBIN="$TEST_ROOT/p4d"
P4BROKERBIN="$TEST_ROOT/broker"
P4TICKETS=/p4/tickets
P4HOME=/p4/blue
LOGS=/p4/blue/logs
P4BROKERPORT=1669
P4PORT=127.0.0.1:1668
[email protected]
GREP=grep
''')
        (common / "backup_functions.sh").write_text('''copy_jd_table () { JDTmpDir="$TEST_ROOT/database"; }
remove_jd_tables () { touch "$TEST_ROOT/cleaned"; }
''')
        self.executable(self.root / "broker", '''[[ ${FAIL_BROKER:-0} == 0 ]] || { echo broker-denied >&2; exit 7; }
printf '%s\n' 'target = localhost:1666;' 'listen = 1667;' 'directory = /p4/broker;' 'tickets=/p4/broker/.p4tickets;' 'logfile = broker.log;' 'admin-email = [email protected];' 'debug-level = server=1;'
''')
        self.executable(self.root / "p4d", '''[[ ${FAIL_QUERY:-0} == 0 ]] || { echo query-denied >&2; exit 8; }
echo 'any: net.autotune = 1'
''')
        path = self.root / "generate.sh"
        path.write_text((BIN / "gen_default_broker_cfg.sh").read_text().replace("/p4/common/bin/", str(common) + "/"))
        return path

    def test_broker_generation_substitutes_endpoint_paths_and_autotune(self):
        script = self.broker_fixture()
        for autotune in (False, True):
            with self.subTest(autotune=autotune):
                if autotune:
                    (self.root / "database/db.config").touch()
                result = self.run_script(script, ["blue"])
                self.assertEqual(result.returncode, 0, result.stderr)
                for expected in ["127.0.0.1:1668", "1669", "/p4/blue/bin", "/p4/tickets", '"/p4/blue/logs/p4broker.log"', '"[email protected]"']:
                    self.assertIn(expected, result.stdout)
                self.assertEqual("net.autotune=1" in result.stdout, autotune)

    def test_broker_failure_never_becomes_successful_empty_configuration(self):
        script = self.broker_fixture()
        result = self.run_script(script, ["blue"], {"FAIL_BROKER": "1"})
        self.assertEqual(result.returncode, 1)
        self.assertEqual(result.stdout, "")
        self.assertIn("broker-denied", result.stderr)

    def test_server_query_failure_cleans_temporary_database_and_stops_generation(self):
        script = self.broker_fixture()
        (self.root / "database/db.config").touch()
        result = self.run_script(script, ["blue"], {"FAIL_QUERY": "1"})
        self.assertEqual(result.returncode, 1)
        self.assertEqual(result.stdout, "")
        self.assertTrue((self.root / "cleaned").exists())

    def test_azure_wrapper_selects_provider_and_preserves_instance_argument(self):
        cloud = SDP / "Server/Unix/p4/common/cloud"
        target = self.root / "cloud/azure/bin/snapshot.sh"
        self.executable(target, (cloud / "azure/bin/snapshot.sh").read_text())
        self.executable(self.root / "cloud/aws/bin/snapshot.sh", 'printf "<%s>\\n" "$@"\nexit 9\n')
        for args, expected in [(["blue"], "blue"), ([], "green")]:
            result = self.run_script(target, args, {"SDP_INSTANCE": "green"})
            self.assertEqual(result.returncode, 9)
            self.assertEqual(result.stdout, "<" + expected + ">\n<azure>\n")

    def test_ec2id_reads_current_cloud_init_link_and_reports_missing_link(self):
        script = self.root / "ec2id"
        link = self.root / "cloud-instance"
        original = SDP / "Server/Unix/p4/common/cloud/aws/bin/ec2id"
        script.write_text(original.read_text().replace("/var/lib/cloud/instance", str(link)))
        link.symlink_to("instances/i-012345abcdef")
        result = self.run_script(script)
        self.assertEqual(result.returncode, 0)
        self.assertEqual(result.stdout, "i-012345abcdef\n")
        link.unlink()
        result = self.run_script(script)
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(result.stdout, "")


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