import io
import json
import tarfile

import pytest

from core.errors import ErrorCode, MatterOSError
from core.io import MAX_INPUT_BYTES, loads
from storage.archive import read_archive
from storage.artifacts import safe_join


def test_path_traversal_blocked(tmp_path):
    assert safe_join(tmp_path, "a/b.txt").parent.parent == tmp_path.resolve()
    for bad in ("../x", "/etc/passwd", "a/../../x"):
        with pytest.raises(MatterOSError):
            safe_join(tmp_path, bad)


def test_artifact_names_cannot_escape(app):
    a = app.artifacts.put_bytes(b"x", "test", "../../evil.txt")
    assert ".." not in a["path"] and a["path"].endswith("evil.txt")


def test_unsafe_archive_members_rejected():
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tf:
        d = b"x"; ti = tarfile.TarInfo("../escape.txt"); ti.size = 1; tf.addfile(ti, io.BytesIO(d))
        m = json.dumps({"members": {}}).encode(); ti = tarfile.TarInfo("manifest.json"); ti.size = len(m); tf.addfile(ti, io.BytesIO(m))
    with pytest.raises(MatterOSError):
        read_archive(buf.getvalue())


def test_parser_limits_and_untrusted_input():
    with pytest.raises(MatterOSError) as e:
        loads("1\n\nH 0 0 0\n" + " " * (MAX_INPUT_BYTES + 1), "xyz")
    assert e.value.code == ErrorCode.RESOURCE_LIMIT
    with pytest.raises(MatterOSError):
        loads("__import__('os').system('echo pwned')", "xyz")


def test_secrets_redacted_from_audit_logs(app):
    app.db.audit("tester", "probe", "x", {"note": "api_key=SUPERSECRET123 password: hunter2"})
    row = app.db.one("SELECT details FROM audit_log WHERE action='probe' ORDER BY id DESC LIMIT 1")
    assert "SUPERSECRET123" not in row["details"] and "hunter2" not in row["details"]


def test_no_shell_tool_and_plugin_allowlist():
    from agents.tools import REGISTRY
    assert not any(k.startswith(("shell", "os.", "fs.", "exec")) for k in REGISTRY)
    from simulation.registry import _load_adapter
    with pytest.raises(ValueError):
        _load_adapter("os:system", None)


def test_paid_services_disabled_by_default(app):
    assert app.config["llm"]["paid_services_enabled"] is False
    assert app.policy["paid_services"]["enabled"] is False and app.policy["cloud_workers"]["enabled"] is False


def test_worker_isolation_cpu_limit_and_cancel_flag_scoped(app, project):
    """Jobs run in a separate process with their own working directory under the jobs dir."""
    from discovery.generators import lj_random_cluster
    c = app.candidates.register(lj_random_cluster(4, 1), project.project_id, "t")
    rec = app.simulation.submit_run(c.candidate_id, "ase_lj", "energy", {})
    j = app.jobs.get(rec.provenance["job_id"])
    assert str(app.jobs.jobs_dir) in j["checkpoint_path"]