# retoor <retoor@molodetz.nl>

import asyncio
import zipfile
import zlib
from pathlib import Path
import pytest
from devplacepy.database import init_db, get_table, refresh_snapshot
from devplacepy import project_files
from devplacepy.project_files import ProjectFileError
from devplacepy.services.jobs import queue
from devplacepy.services.jobs.zip_service import ZipService
from devplacepy.services.jobs import zip_worker
from tests.conftest import run_async
@pytest.fixture(autouse=True)
def _init_db_zip_jobs():
    init_db()
    yield
@pytest.fixture
def zip_env(tmp_path, monkeypatch):
    monkeypatch.setattr(
        "devplacepy.services.jobs.zip_service.ZIPS_DIR", tmp_path / "zips"
    )
    monkeypatch.setattr(
        "devplacepy.services.jobs.zip_service.ZIP_STAGING_DIR", tmp_path / "staging"
    )
    monkeypatch.setattr("devplacepy.project_files.PROJECT_FILES_DIR", tmp_path / "pf")
    yield tmp_path
    jobs = get_table("jobs")
    for row in list(jobs.find(kind="zip")):
        jobs.delete(uid=row["uid"])
    files = get_table("project_files")
    for row in list(files.find()):
        if str(row.get("project_uid", "")).startswith("ziptest"):
            files.delete(uid=row["uid"])
_pid_counter = [0]
def _make_project(text=None, binary=False):
    _pid_counter[0] += 1
    pid = f"ziptest-{_pid_counter[0]}"
    user = {"uid": f"ziptest-owner-{_pid_counter[0]}"}
    project_files.write_text_file(pid, user, "README.md", "# hello\nworld")
    project_files.write_text_file(pid, user, "src/app.py", "print(1)\n")
    if binary:
        project_files.store_upload(pid, user, "assets", "logo.bin", bytes(range(256)))
    return pid, user
def _process_zip_jobs():
    async def drive():
        svc = ZipService()
        for _ in range(400):
            await svc.run_once()
            refresh_snapshot()
            pending = [
                r
                for r in get_table("jobs").find(kind="zip")
                if r["status"] in ("pending", "running")
            ]
            if not pending and not svc._inflight:
                return
            await asyncio.sleep(0.05)

    run_async(drive())


def test_enqueue_creates_pending_job(zip_env):
    uid = queue.enqueue("zip", {"a": 1}, "user", "u1", "Name")
    job = queue.get_job(uid)
    assert job["status"] == "pending"
    assert job["kind"] == "zip"
    assert job["payload"] == {"a": 1}
    assert job["result"] == {}
    assert job["owner_kind"] == "user" and job["owner_id"] == "u1"
    assert job["created_at"] and job["expires_at"] == ""


def test_get_job_missing_returns_none(zip_env):
    assert queue.get_job("does-not-exist") is None


def test_touch_job_extends_expiry(zip_env):
    uid = queue.enqueue("zip", {}, "guest", "g1", "x")
    queue.touch_job(uid, 3600)
    job = queue.get_job(uid)
    assert job["last_accessed_at"] and job["expires_at"]


def test_list_jobs_filters(zip_env):
    a = queue.enqueue("zip", {}, "user", "owner-A", "a")
    queue.enqueue("zip", {}, "user", "owner-B", "b")
    only_a = queue.list_jobs(kind="zip", owner=("user", "owner-A"))
    assert [j["uid"] for j in only_a] == [a]


def test_zip_worker_stats_and_crc(tmp_path):
    src = tmp_path / "src"
    (src / "sub").mkdir(parents=True)
    (src / "a.txt").write_text("hello")
    (src / "sub" / "b.txt").write_text("world")
    out = tmp_path / "out.zip"
    stats = zip_worker._build(str(src), str(out))
    assert stats["file_count"] == 2
    assert stats["dir_count"] == 1
    assert stats["bytes_in"] == 10
    assert stats["bytes_out"] == out.stat().st_size
    expected_crc = 0
    with open(out, "rb") as handle:
        for chunk in iter(lambda: handle.read(65536), b""):
            expected_crc = zlib.crc32(chunk, expected_crc)
    assert stats["crc32"] == (expected_crc & 0xFFFFFFFF)
    assert sorted(zipfile.ZipFile(out).namelist()) == ["a.txt", "sub/", "sub/b.txt"]


def test_process_whole_project(zip_env):
    pid, _ = _make_project(binary=True)
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": pid, "path": ""}},
        "user",
        "u",
        "Proj",
    )
    _process_zip_jobs()
    job = queue.get_job(uid)
    assert job["status"] == "done"
    assert job["duration_ms"] >= 0
    assert job["expires_at"]
    result = job["result"]
    names = sorted(zipfile.ZipFile(result["local_path"]).namelist())
    assert names == ["README.md", "assets/", "assets/logo.bin", "src/", "src/app.py"]
    assert result["file_count"] == 3 and result["dir_count"] == 2
    assert job["bytes_out"] == Path(result["local_path"]).stat().st_size


def test_zip_with_missing_blob_succeeds(zip_env):
    pid, _ = _make_project(binary=True)
    binary_row = get_table("project_files").find_one(
        project_uid=pid, is_binary=1
    )
    assert binary_row is not None
    blob_path = project_files.PROJECT_FILES_DIR / binary_row["directory"] / binary_row["stored_name"]
    assert blob_path.is_file()
    blob_path.unlink()
    assert not blob_path.exists()
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": pid, "path": ""}},
        "user",
        "u",
        "ProjMissing",
    )
    _process_zip_jobs()
    job = queue.get_job(uid)
    assert job["status"] == "done"
    assert job["bytes_out"] > 0
    assert job["result"]["file_count"] == 2
    names = sorted(zipfile.ZipFile(job["result"]["local_path"]).namelist())
    assert "assets/logo.bin" not in names


def test_process_subtree_folder(zip_env):
    pid, _ = _make_project()
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": pid, "path": "src"}},
        "user",
        "u",
        "src",
    )
    _process_zip_jobs()
    result = queue.get_job(uid)["result"]
    assert sorted(zipfile.ZipFile(result["local_path"]).namelist()) == [
        "src/",
        "src/app.py",
    ]


def test_process_single_file(zip_env):
    pid, _ = _make_project()
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": pid, "path": "src/app.py"}},
        "user",
        "u",
        "app.py",
    )
    _process_zip_jobs()
    result = queue.get_job(uid)["result"]
    assert zipfile.ZipFile(result["local_path"]).namelist() == ["app.py"]


def test_final_name_format_and_strip_zip(zip_env):
    pid, _ = _make_project()
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": pid, "path": ""}},
        "user",
        "u",
        "My Project.ZIP",
    )
    _process_zip_jobs()
    result = queue.get_job(uid)["result"]
    name = result["final_name"]
    assert name.endswith(".my-project.zip")
    crc_part = name.split(".", 1)[0]
    assert len(crc_part) == 8 and int(crc_part, 16) >= 0
    assert crc_part == result["crc32"]


def test_blank_preferred_name_defaults(zip_env):
    pid, _ = _make_project()
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": pid, "path": ""}},
        "user",
        "u",
        "",
    )
    _process_zip_jobs()
    assert queue.get_job(uid)["result"]["final_name"].endswith(".download.zip")


def test_identical_content_replaces_same_name(zip_env):
    pid, _ = _make_project()
    payload = {"source": {"type": "project_tree", "project_uid": pid, "path": ""}}
    uid1 = queue.enqueue("zip", payload, "user", "u", "thing")
    _process_zip_jobs()
    name1 = queue.get_job(uid1)["result"]["final_name"]
    uid2 = queue.enqueue("zip", payload, "user", "u", "thing")
    _process_zip_jobs()
    name2 = queue.get_job(uid2)["result"]["final_name"]
    assert name1 == name2


def test_cleanup_removes_artifact(zip_env):
    pid, _ = _make_project()
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": pid, "path": ""}},
        "user",
        "u",
        "thing",
    )
    _process_zip_jobs()
    job = queue.get_job(uid)
    path = Path(job["result"]["local_path"])
    assert path.is_file()
    ZipService().cleanup(job)
    assert not path.exists()


def test_unsupported_source_fails_job(zip_env):
    uid = queue.enqueue("zip", {"source": {"type": "evil"}}, "user", "u", "x")
    _process_zip_jobs()
    job = queue.get_job(uid)
    assert job["status"] == "failed"
    assert "unsupported" in job["error"].lower()


def test_normalize_path_rejects_parent(zip_env):
    with pytest.raises(ProjectFileError):
        project_files.normalize_path("a/../../b")
    with pytest.raises(ProjectFileError):
        project_files.normalize_path("../etc/passwd")


def test_export_blocks_malicious_db_path(zip_env, tmp_path):
    pid = "ziptest-evil"
    get_table("project_files").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": "evil-node",
            "project_uid": pid,
            "user_uid": "u",
            "path": "../escape.txt",
            "name": "escape.txt",
            "parent_path": "",
            "type": "file",
            "content": "pwned",
            "is_binary": 0,
            "stored_name": None,
            "directory": None,
            "mime_type": "text/plain",
            "size": 5,
            "created_at": "x",
            "updated_at": "x",
        }
    )
    dest = tmp_path / "dest"
    with pytest.raises(ProjectFileError):
        project_files.export_to_dir(pid, "", dest)
    assert not (tmp_path / "escape.txt").exists()


def test_traversal_payload_marks_job_failed(zip_env):
    pid = "ziptest-evil2"
    get_table("project_files").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": "evil-node2",
            "project_uid": pid,
            "user_uid": "u",
            "path": "../../escape2.txt",
            "name": "escape2.txt",
            "parent_path": "",
            "type": "file",
            "content": "pwned",
            "is_binary": 0,
            "stored_name": None,
            "directory": None,
            "mime_type": "text/plain",
            "size": 6,
            "created_at": "x",
            "updated_at": "x",
        }
    )
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": pid, "path": ""}},
        "user",
        "u",
        "evil",
    )
    _process_zip_jobs()
    assert queue.get_job(uid)["status"] == "failed"


def test_orphan_running_recovered_on_enable(zip_env):
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": "p", "path": ""}},
        "user",
        "u",
        "n",
    )
    get_table("jobs").update(
        {"uid": uid, "status": "running", "started_at": "2020-01-01T00:00:00+00:00"},
        ["uid"],
    )
    svc = ZipService()
    run_async(svc.on_enable())
    job = queue.get_job(uid)
    assert job["status"] == "pending"
    assert job["retry_count"] == 1


def test_orphan_exceeds_retry_limit_fails(zip_env):
    uid = queue.enqueue(
        "zip",
        {"source": {"type": "project_tree", "project_uid": "p", "path": ""}},
        "user",
        "u",
        "n",
    )
    get_table("jobs").update(
        {
            "uid": uid,
            "status": "running",
            "retry_count": 3,
            "started_at": "2020-01-01T00:00:00+00:00",
        },
        ["uid"],
    )
    svc = ZipService()
    run_async(svc.on_enable())
    assert queue.get_job(uid)["status"] == "failed"


def test_retention_sweep_deletes_expired(zip_env, tmp_path):
    artifact = tmp_path / "old.zip"
    artifact.write_bytes(b"PK\x05\x06" + b"\x00" * 18)
    uid = queue.enqueue("zip", {}, "user", "u", "old")
    get_table("jobs").update(
        {
            "uid": uid,
            "status": "done",
            "result": '{"local_path": "%s"}' % artifact.as_posix(),
            "expires_at": "2000-01-01T00:00:00+00:00",
        },
        ["uid"],
    )
    svc = ZipService()
    run_async(svc.run_once())
    refresh_snapshot()
    assert queue.get_job(uid) is None
    assert not artifact.exists()


def test_retention_keeps_unexpired(zip_env, tmp_path):
    uid = queue.enqueue("zip", {}, "user", "u", "fresh")
    get_table("jobs").update(
        {
            "uid": uid,
            "status": "done",
            "result": "{}",
            "expires_at": "2999-01-01T00:00:00+00:00",
        },
        ["uid"],
    )
    svc = ZipService()
    run_async(svc.run_once())
    refresh_snapshot()
    assert queue.get_job(uid) is not None


def test_max_concurrent_caps_inflight(zip_env, monkeypatch):
    for _ in range(4):
        queue.enqueue(
            "zip",
            {"source": {"type": "project_tree", "project_uid": "p", "path": ""}},
            "user",
            "u",
            "n",
        )

    async def fake_process(self, job):
        await asyncio.sleep(0.2)
        return {}

    monkeypatch.setattr(ZipService, "process", fake_process)
    monkeypatch.setattr(ZipService, "max_concurrent", lambda self: 2)

    async def check():
        svc = ZipService()
        svc._refill()
        count = len(svc._inflight)
        for entry in svc._inflight.values():
            entry["task"].cancel()
        return count

    assert run_async(check()) == 2
