# FILE: _St/AutoDistribute/test_autodistribute.py | ROLE: 개인 배포 build/ZIP/SFTP 회귀 테스트 from __future__ import annotations import hashlib import json import subprocess import tempfile import zipfile from pathlib import Path from AutoDistribute import ( TARGET_DIR_NAMES, BuildError, UploadError, VersionMeta, build_st_zip, build_winscp_script, choose_next_build, load_sftp_config, parse_env_conf, parse_version_html, render_version_html, resolve_st_root, upload_zip, validate_zip, ) def _write(path: Path, text: str = "x") -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") def _layout(root: Path) -> tuple[Path, Path]: st = root / "_St" tool = st / "AutoDistribute" tool.mkdir(parents=True) for name in TARGET_DIR_NAMES: _write(st / name / "keep.txt", name) manifest = st / "IG_StYellow" / "manifest.json" manifest.write_text(json.dumps({"version": "2.3.8.3200", "version_name": "2.3.8.320"}), encoding="utf-8") return st, tool def _conf(tool: Path) -> Path: conf = tool / ".env.conf" _write( conf, "\n".join( [ "# test", "SFTP_ID=root", "SFTP_PW=p!@ss", "WINSCP_EXE=C:\\Program Files (x86)\\WinSCP\\WinSCP.com", 'WINSCP_OPEN=open sftp://{SFTP_ID_URL}:{SFTP_PW_URL}@server:2022/ -hostkey="ssh-ed25519 255 TEST" -rawsettings AuthGSSAPI=0 FSProtocol=2', "REMOTE_FILE=/var/www/html/aiwk/down/_St.zip", "REMOTE_VERSION_FILE=/var/www/html/aiwk/down/_St_ver.html", "VERSION_URL=http://server/down/_St_ver.html", "WINSCP_LOG=WinSCP.log", "", ] ), ) return conf def test_fixed_seven_folders_plus_root_version_only() -> None: with tempfile.TemporaryDirectory() as td: st, tool = _layout(Path(td)) _write(st / "OtherModule" / "must_not_ship.txt") _conf(tool) output = tool / "_St.zip" stats = build_st_zip(st_root=st, output=output, base_dir=tool) with zipfile.ZipFile(output) as zf: names = set(zf.namelist()) assert "_St/_St_ver.html" in names assert "_St/OtherModule/must_not_ship.txt" not in names assert all(name == "_St/_St_ver.html" or name.split("/")[1] in TARGET_DIR_NAMES for name in names) assert stats.version == "2.3.8.320" assert stats.build >= 320 def test_excludes_read_logs_data_and_python_cache() -> None: with tempfile.TemporaryDirectory() as td: st, tool = _layout(Path(td)) _conf(tool) target = st / "Auto8700" _write(target / "ai_read.md") _write(target / "read_note.txt") _write(target / "logs" / "run.txt") _write(target / "data" / "state.sqlite3") _write(target / "__pycache__" / "x.pyc") _write(target / "temp.pyc") output = tool / "_St.zip" build_st_zip(st_root=st, output=output, base_dir=tool) _, violations = validate_zip(output) assert violations == [] with zipfile.ZipFile(output) as zf: text = "\n".join(zf.namelist()).casefold() assert "ai_read" not in text and "read_note" not in text assert "/logs/" not in text and "/data/" not in text and "__pycache__" not in text def test_missing_fixed_folder_is_fail() -> None: with tempfile.TemporaryDirectory() as td: st, tool = _layout(Path(td)) (st / "WS8771" / "keep.txt").unlink() (st / "WS8771").rmdir() try: resolve_st_root(tool) except BuildError as exc: assert "WS8771" in str(exc) else: raise AssertionError("missing folder must fail") def test_build_is_integer_and_never_goes_backward() -> None: with tempfile.TemporaryDirectory() as td: st, _tool = _layout(Path(td)) (st / "_St_ver.html").write_text(render_version_html(VersionMeta("old", 450)), encoding="utf-8") meta = choose_next_build( st, version_url="http://server/_St_ver.html", remote_getter=lambda _url, _timeout: VersionMeta("server", 500), ) assert isinstance(meta.build, int) assert meta.build == 501 assert meta.version == "2.3.8.320" def test_version_html_roundtrip() -> None: meta = VersionMeta("2.3.8.320", 777, "_St.zip", "a" * 64) parsed = parse_version_html(render_version_html(meta)) assert parsed == meta def test_zip_server_meta_sha256_matches_created_zip_and_local_meta_has_same_build() -> None: with tempfile.TemporaryDirectory() as td: st, tool = _layout(Path(td)) _conf(tool) stats = build_st_zip(st_root=st, output=tool / "_St.zip", base_dir=tool) assert stats.sha256 == hashlib.sha256(stats.output.read_bytes()).hexdigest() local = parse_version_html((st / "_St_ver.html").read_text(encoding="utf-8")) server = parse_version_html((tool / "_St_ver.html").read_text(encoding="utf-8")) assert local.build == server.build == stats.build assert local.version == server.version == stats.version assert local.sha256 == "" assert server.sha256 == stats.sha256 def test_env_conf_and_open_command_expansion() -> None: with tempfile.TemporaryDirectory() as td: tool = Path(td) conf = _conf(tool) raw = parse_env_conf(conf) assert raw["SFTP_ID"] == "root" config = load_sftp_config(conf, base_dir=tool, require_exe=False) assert "root:p%21%40ss@server:2022/" in config.open_command assert config.remote_file == "/var/www/html/aiwk/down/_St.zip" assert config.remote_version_file == "/var/www/html/aiwk/down/_St_ver.html" assert config.version_url == "http://server/down/_St_ver.html" def test_winscp_script_uploads_zip_first_and_version_meta_last() -> None: with tempfile.TemporaryDirectory() as td: tool = Path(td) conf = _conf(tool) _write(tool / "_St.zip") _write(tool / "_St_ver.html") config = load_sftp_config(conf, base_dir=tool, require_exe=False) script = build_winscp_script(config, tool / "_St.zip", tool / "_St_ver.html") zip_pos = script.index('/var/www/html/aiwk/down/_St.zip') ver_pos = script.index('/var/www/html/aiwk/down/_St_ver.html') assert zip_pos < ver_pos assert script.rstrip().endswith("exit") def test_upload_runs_winscp_and_sends_both_files() -> None: with tempfile.TemporaryDirectory() as td: st, tool = _layout(Path(td)) conf = _conf(tool) output = tool / "_St.zip" build_st_zip(st_root=st, output=output, base_dir=tool) calls: list[tuple[list[str], dict[str, object]]] = [] def fake_runner(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: calls.append((cmd, kwargs)) script_arg = next(item for item in cmd if item.startswith("/script=")) text = Path(script_arg.split("=", 1)[1]).read_text(encoding="utf-8-sig") assert text.count("put ") == 2 assert "/var/www/html/aiwk/down/_St.zip" in text assert "/var/www/html/aiwk/down/_St_ver.html" in text return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="") stats = upload_zip(output, conf_path=conf, base_dir=tool, runner=fake_runner) assert stats.returncode == 0 assert len(calls) == 1 assert not list(tool.glob("autodistribute_winscp_*.txt")) def test_upload_nonzero_exit_is_fail() -> None: with tempfile.TemporaryDirectory() as td: st, tool = _layout(Path(td)) conf = _conf(tool) output = tool / "_St.zip" build_st_zip(st_root=st, output=output, base_dir=tool) def fake_runner(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="network fail") try: upload_zip(output, conf_path=conf, base_dir=tool, runner=fake_runner) except UploadError as exc: assert "exit=1" in str(exc) else: raise AssertionError("nonzero WinSCP exit must fail") assert not list(tool.glob("autodistribute_winscp_*.txt"))