from pathlib import Path import pytest from deployer import config, detect, pipeline, remote from deployer.kinds import get SERVER = """ [server] host = "h" app_root = "/srv/app" """ def write(tmp_path: Path, body: str) -> Path: p = tmp_path / "deploy.toml" p.write_text(SERVER + body) return p def test_load_and_dest_paths(tmp_path): cfg = config.load(write(tmp_path, """ [[app]] name = "a" kind = "python-update" src = "~/x" dest = "python/a" unit = "a" configs = ["configs"] [[app]] name = "b" kind = "files" src = "/tmp" dest = "~/scripts" """)) assert [a.name for a in cfg.apps] == ["a", "b"] assert cfg.dest_path(cfg.app("a")) == "/srv/app/python/a" assert cfg.dest_path(cfg.app("b")) == "/home/user/scripts" assert cfg.app("a").src_path == Path.home() / "x" @pytest.mark.parametrize("body, msg", [ ('name = "a"\nkind = "nope"\nsrc = "x"\ndest = "y"', "kind must be"), ('name = "a"\nkind = "python-update"\nsrc = "x"\ndest = "y"', "unit"), ('name = "a"\nkind = "files"\nsrc = "x"\ndest = "../etc"', "unsafe dest"), ('name = "a"\nkind = "files"\nsrc = "x"\ndest = "y"\ndelete = "yes"', "wrong type"), ('name = "a"\nkind = "files"\nsrc = "x"\ndest = "y"\ntypo = 1', "unknown key"), ('name = "a"\nkind = "artifact"\nsrc = "x"\ndest = "y"', "artifact"), ('name = "a"\nkind = "files"\nsrc = "x"\ndest = "y"\nconfigs = ["a/b"]', "top-level"), ]) def test_validation(tmp_path, body, msg): with pytest.raises(config.ConfigError, match=msg): config.load(write(tmp_path, "[[app]]\n" + body)) def test_append_round_trip(tmp_path): p = write(tmp_path, "# keep me\n") app = config.App(name="n", kind="docker", src="~/p", dest="docker/n", build_env={"JAVA_HOME": "~/jdk"}, post='echo "hi"') draft = config.App(name="d", kind="python-update", src="", dest="python/d", unit="d") config.append_apps(p, [app], [draft]) text = p.read_text() assert "# keep me" in text and '# name = "d"' in text cfg = config.load(p) assert [a.name for a in cfg.apps] == ["n"] assert cfg.app("n") == app def test_filters_protect_server_items(): rules = remote.build_filters(protected=["configs", "Makefile"], exclude=["/tests"], use_gitignore=True) assert rules[:2] == ["- /Makefile", "- /configs"] assert "- .venv" in rules and "- /tests" in rules assert rules[-1] == ":- .gitignore" assert ":- .gitignore" not in remote.build_filters(protected=[], exclude=[], use_gitignore=False) def test_rsync_cmd_options(): s = config.Server() cmd = remote.rsync_cmd(s, "src/", "/d/", filters=["- x"], delete=True, dry_run=True, extra=["--chmod=F640"]) assert cmd[0] == "rsync" and "-n" in cmd and "--delete" in cmd and "-c" in cmd assert cmd.index("--chmod=F640") < cmd.index("--filter=- x") assert cmd[-2:] == ["src/", "user@localhost:/d/"] def test_parse_itemize(): out = "\n".join([ ".d..t...... ./", " target", "created 2 directories for /x", ]) got = [(c.action, c.path) for c in remote.parse_itemize(out)] assert got == [("new", "new.py"), ("update", "changed.py"), ("new", "pkg/"), ("delete", "old.py"), ("new", "link -> target")] @pytest.mark.parametrize("local, deployed, state", [ ("abc1234", "abc1234", "same"), ("def5678", "abc1234", "changed"), ("def5678-dirty", "abc1234", "changed"), ("abc1234-dirty", "abc1234", "dirty"), ("abc1234", "abc1234-dirty", "dirty"), ("", "abc1234", "?"), ("abc1234", "", "?"), ("abc1234", "-", "?"), ]) def test_code_state(local, deployed, state): assert pipeline.code_state(local, deployed) == state def test_deployed_rev(): line = "2026-10-17T10:57:30 EO_Service {} python-update {} rev=22fc8b3 snap=20261017-105714" assert pipeline.deployed_rev(line.format("deploy", "ok").split()) == "22fc8b3" assert pipeline.deployed_rev(line.format("deploy", "UNHEALTHY").split()) == "22fc8b3" assert pipeline.deployed_rev(line.format("deploy", "FAILED").split()) == "" assert pipeline.deployed_rev(line.format("rollback", "ok").split()) == "" def test_shell_vars_quote(): v = remote.shell_vars(A="it's", L=["x y", "$z"]) assert v == "A='it'\"'\"'s'\nL=('x y' '$z')\n" def test_detect_kinds(tmp_path): py = tmp_path / "MyApp" (py / "configs").mkdir(parents=True) (py / "app.py").write_text("") (py / "secrets.yaml").write_text("") (py / "docker-compose.yml").write_text("") app = detect.new_app(py) assert (app.kind, app.dest, app.unit, app.entry) == ("python-service", "python/MyApp", "MyApp", "app.py") assert app.configs == ["configs", "secrets.yaml"] dk = tmp_path / "stack" dk.mkdir() (dk / "compose.yaml").write_text("") assert detect.new_app(dk).kind == "docker" jv = tmp_path / "Svc" jv.mkdir() (jv / "pom.xml").write_text("war") (jv / "mvnw").write_text("") a = detect.new_app(jv) assert a.kind == "artifact" and a.artifact == "target/*.war" and a.build.startswith("./mvnw") def test_match_local(tmp_path): a = tmp_path / "DasSimpleExporter" a.mkdir() (a / "dasExporter.service").write_text("") b = tmp_path / "weather-proxy" b.mkdir() assert detect.match_local([a, b], ["WeatherProxy"]) == b assert detect.match_local([a, b], ["DasExporter"], unit="dasExporter") == a assert detect.match_local([a, b], ["nothing"]) is None def test_python_config_layout(tmp_path): (tmp_path / "configs").mkdir() cfg = config.Config(tmp_path / "x.toml", config.Server(), []) plan = type("P", (), {})() plan.cfg, plan.dest = cfg, "/home/user/app/python/App" plan.app = config.App(name="App", kind="python-update", src=str(tmp_path), dest="python/App", unit="u", configs=["configs"]) k = get("python-update") assert k.config_items(plan) == [("configs", "/home/user/app/configs/python/App")] plan.app.configs = [".env", "secrets.yaml"] assert k.config_items(plan) == [(".env", "/home/user/app/configs/python/App/app.env"), ("secrets.yaml", "/home/user/app/configs/python/App/secrets.yaml")] plan.dest = "/home/user/app/docker/stack" plan.app = config.App(name="stack", kind="docker", src=str(tmp_path), dest="docker/stack") assert get("docker").config_items(plan) == [(".env", "/home/user/app/configs/docker/stack.env")]