104 lines
4.8 KiB
Python
104 lines
4.8 KiB
Python
import json
|
||
import os
|
||
import stat
|
||
|
||
import pytest
|
||
|
||
from jira_hub import release
|
||
|
||
|
||
def agent_file(tmp_path, version, body="print('x')\n", name="agent.py"):
|
||
p = tmp_path / name
|
||
p.write_text(f'#!/usr/bin/env python3\n"""агент"""\nAGENT_VERSION = {version}\n{body}', encoding="utf-8")
|
||
return p
|
||
|
||
|
||
def test_publish_writes_file_and_manifest(tmp_path):
|
||
rd = tmp_path / "release"
|
||
src = agent_file(tmp_path, 6)
|
||
m = release.publish(rd, src)
|
||
assert m["version"] == 6 and m["size"] == src.stat().st_size and len(m["sha256"]) == 64
|
||
assert (rd / "agent.py").read_bytes() == src.read_bytes()
|
||
assert json.loads((rd / "manifest.json").read_text(encoding="utf-8")) == m
|
||
assert stat.S_IMODE(os.stat(rd).st_mode) == 0o700
|
||
assert release.current(rd) == {"version": 6, "sha256": m["sha256"], "published_at": m["published_at"]}
|
||
loaded = release.load(rd)
|
||
assert loaded == {"version": 6, "sha256": m["sha256"], "content": src.read_text(encoding="utf-8")}
|
||
assert not [p.name for p in rd.iterdir() if p.name.startswith(".")] # временных файлов не осталось
|
||
|
||
|
||
def test_publish_only_up(tmp_path):
|
||
rd = tmp_path / "release"
|
||
release.publish(rd, agent_file(tmp_path, 6))
|
||
before = (rd / "manifest.json").read_bytes()
|
||
with pytest.raises(release.PublishError, match="опубликована v6, новая должна быть больше"):
|
||
release.publish(rd, agent_file(tmp_path, 6, body="print('y')\n"))
|
||
with pytest.raises(release.PublishError, match="опубликована v6"):
|
||
release.publish(rd, agent_file(tmp_path, 5))
|
||
assert (rd / "manifest.json").read_bytes() == before
|
||
assert release.publish(rd, agent_file(tmp_path, 7))["version"] == 7
|
||
|
||
|
||
@pytest.mark.parametrize("text", ["print(1)\n", "AGENT_VERSION = 5\nAGENT_VERSION = 6\n", "AGENT_VERSION = 'v5'\n",
|
||
" AGENT_VERSION = 5\n"])
|
||
def test_version_line_required_once(tmp_path, text):
|
||
p = tmp_path / "a.py"
|
||
p.write_text(text, encoding="utf-8")
|
||
with pytest.raises(release.PublishError, match="AGENT_VERSION"):
|
||
release.publish(tmp_path / "release", p)
|
||
assert not (tmp_path / "release" / "manifest.json").exists()
|
||
|
||
|
||
def test_publish_rejects_non_utf8_and_big(tmp_path):
|
||
p = tmp_path / "a.py"
|
||
p.write_bytes(b"AGENT_VERSION = 5\n\xff\xfe")
|
||
with pytest.raises(release.PublishError, match="UTF-8"):
|
||
release.publish(tmp_path / "release", p)
|
||
p.write_bytes(b"AGENT_VERSION = 5\n" + b"#" * (1024 * 1024))
|
||
with pytest.raises(release.PublishError, match="1 МБ"):
|
||
release.publish(tmp_path / "release", p)
|
||
with pytest.raises(release.PublishError, match="не прочитан"):
|
||
release.publish(tmp_path / "release", tmp_path / "нет.py")
|
||
|
||
|
||
def test_current_none_without_release(tmp_path):
|
||
assert release.current(tmp_path / "release") is None
|
||
assert release.load(tmp_path / "release") is None
|
||
|
||
|
||
def test_current_broken_manifest(tmp_path):
|
||
rd = tmp_path / "release"
|
||
rd.mkdir()
|
||
(rd / "manifest.json").write_text('{"version": 6, "sha', encoding="utf-8")
|
||
assert release.current(rd) is None
|
||
(rd / "manifest.json").write_text('{"version": "шесть", "sha256": "x"}', encoding="utf-8")
|
||
assert release.current(rd) is None
|
||
|
||
|
||
def test_load_rejects_sha_mismatch(tmp_path):
|
||
rd = tmp_path / "release"
|
||
release.publish(rd, agent_file(tmp_path, 6))
|
||
(rd / "agent.py").write_text("AGENT_VERSION = 6\nподменён\n", encoding="utf-8")
|
||
assert release.current(rd)["version"] == 6
|
||
assert release.load(rd) is None
|
||
|
||
|
||
def test_hint():
|
||
rel = {"version": 6, "sha256": "s", "published_at": "t"}
|
||
online5 = {"agent_version": "5", "online": True}
|
||
assert release.hint(rel, online5, None, None) == "агент запущен без run.py — обновится только вручную"
|
||
assert release.hint(rel, online5, None, 1) is None
|
||
assert release.hint(rel, {"agent_version": "5", "online": False}, None, None) is None
|
||
assert release.hint(rel, {"agent_version": "6", "online": True}, None, None) is None
|
||
assert release.hint(None, online5, None, None) is None
|
||
rb = {"state": "rolled_back", "from_version": 6, "to_version": 7, "error": "код 1 через 3 с"}
|
||
assert release.hint(rel, {"agent_version": "6", "online": True}, rb, 1) == "откат с v7 на v6: код 1 через 3 с"
|
||
rj = {"state": "rejected", "from_version": 5, "to_version": 6, "error": "sha256 не сходится"}
|
||
assert release.hint(rel, online5, rj, 1) == "v6 отвергнута: sha256 не сходится"
|
||
assert release.hint(rel, online5, {"state": "ok", "from_version": 5, "to_version": 6}, 1) is None
|
||
|
||
|
||
def test_hint_none_when_no_agent():
|
||
rel = {"version": 6, "sha256": "s", "published_at": "t"}
|
||
assert release.hint(rel, {"agent_version": None, "online": False}, None, None) is None
|