Blame for tests/test_sysinfo.py

import pytest

from bot import sysinfo


def test_parse_cpu_max():
    assert sysinfo.parse_cpu_max("200000 100000") == 2.0
    assert sysinfo.parse_cpu_max("max 100000") is None
    assert sysinfo.parse_cpu_max("garbage") is None
    assert sysinfo.parse_cpu_max(None) is None


def test_parse_mem_limit():
    assert sysinfo.parse_mem_limit("17179869184") == 17179869184
    assert sysinfo.parse_mem_limit("max") is None
    assert sysinfo.parse_mem_limit("9223372036854771712") is None  # cgroup v1 "unlimited"
    assert sysinfo.parse_mem_limit(None) is None


def test_parse_stat_value():
    text = "usage_usec 12345\nuser_usec 1\ninactive_file 99"
    assert sysinfo.parse_stat_value(text, "usage_usec") == 12345
    assert sysinfo.parse_stat_value(text, "inactive_file") == 99
    assert sysinfo.parse_stat_value(text, "missing") is None


@pytest.fixture
def fake_cgroup(tmp_path, monkeypatch):
    (tmp_path / "cpu.max").write_text("200000 100000\n")
    (tmp_path / "memory.max").write_text("17179869184\n")
    (tmp_path / "memory.current").write_text("3000000000\n")
    (tmp_path / "memory.stat").write_text("anon 1\ninactive_file 1000000000\n")
    (tmp_path / "cpu.stat").write_text("usage_usec 1000000\n")
    monkeypatch.setattr(sysinfo, "CGROUP_ROOT", str(tmp_path))
    return tmp_path


def test_container_memory_uses_cgroup_and_ignores_page_cache(fake_cgroup):
    used, total = sysinfo.memory_usage()
    assert used == 2_000_000_000
    assert total <= 17179869184
    assert sysinfo.cpu_cores() == 2.0


async def test_container_cpu_percent_from_cgroup(fake_cgroup, monkeypatch):
    calls = {"n": 0}
    real = sysinfo._cgroup_cpu_usec

    def fake_usec():
        calls["n"] += 1
        return 1_000_000 if calls["n"] == 1 else 1_500_000  # 0.5 cpu-seconds over ~0.1s window

    monkeypatch.setattr(sysinfo, "_cgroup_cpu_usec", fake_usec)
    pct = await sysinfo.cpu_percent(interval=0.1)
    assert 0 < pct <= 100
    monkeypatch.setattr(sysinfo, "_cgroup_cpu_usec", real)


def test_falls_back_to_psutil_without_cgroup(tmp_path, monkeypatch):
    monkeypatch.setattr(sysinfo, "CGROUP_ROOT", str(tmp_path / "does-not-exist"))
    used, total = sysinfo.memory_usage()
    assert total > 0 and used >= 0
    assert sysinfo.cpu_cores() >= 1


async def test_format_status_and_stats(tmp_path):
    text = sysinfo.format_status(123.4, 1, "1.45.0")
    assert "123 ms" in text and "DC 1" in text and "v1.0.0" in text and "Uptime" in text
    assert "n/a" in sysinfo.format_status(None, None, "x")
    stats = await sysinfo.format_stats(3, 12, 4_500_000_000, 1, 2, str(tmp_path))
    for needle in ("Users:</b> 3", "Files renamed", "1 running · 2 waiting", "CPU", "RAM", "Disk", "Network"):
        assert needle in stats


# ---------------------------------------------------------------------------
# what makes transfers fast: cryptg, uvloop, parallel transfer
# ---------------------------------------------------------------------------


def test_telethon_encrypts_with_cryptg_here():
    # also a guard for the deployment: requirements.txt must keep installing cryptg
    assert sysinfo.crypto_backend() == "cryptg"


@pytest.mark.parametrize("cryptg, libssl_works, expected", [
    (object(), True, "cryptg"), (object(), False, "cryptg"), (None, True, "libssl"), (None, False, "python"),
])
def test_crypto_backend_follows_the_order_telethon_picks(monkeypatch, cryptg, libssl_works, expected):
    from telethon.crypto import aes

    monkeypatch.setattr(aes, "cryptg", cryptg)
    monkeypatch.setattr(aes.libssl, "encrypt_ige", (lambda *a: b"") if libssl_works else None)
    monkeypatch.setattr(aes.libssl, "decrypt_ige", (lambda *a: b"") if libssl_works else None)
    assert sysinfo.crypto_backend() == expected


async def test_loop_backend_sees_the_standard_loop():
    assert sysinfo.loop_backend() == "asyncio"


def test_loop_backend_outside_a_loop_and_on_uvloop():
    uvloop = pytest.importorskip("uvloop")
    assert sysinfo.loop_backend() == "none"
    loop = uvloop.new_event_loop()
    try:
        assert loop.run_until_complete(_loop_name()) == "uvloop"
    finally:
        loop.close()


async def _loop_name():
    return sysinfo.loop_backend()


def test_the_bot_really_starts_on_uvloop():
    from bot import main as bot_main

    pytest.importorskip("uvloop")
    seen = []

    async def probe():
        seen.append(sysinfo.loop_backend())
        return 0

    assert bot_main._run(probe) == 0 and seen == ["uvloop"]


def test_speed_summary_and_checks_mark_what_is_off(monkeypatch):
    monkeypatch.setattr(sysinfo, "crypto_backend", lambda: "cryptg")
    monkeypatch.setattr(sysinfo, "loop_backend", lambda: "uvloop")
    assert sysinfo.speed_summary(True, 8) == "cryptg ✅ · uvloop ✅ · parallel ×8 ✅"
    assert all(fine for _short, _detail, fine in sysinfo.speed_checks(True, 8))

    monkeypatch.setattr(sysinfo, "crypto_backend", lambda: "python")
    monkeypatch.setattr(sysinfo, "loop_backend", lambda: "asyncio")
    assert sysinfo.speed_summary(False, 8) == "slow Python crypto ⚠️ · no uvloop ⚠️ · parallel off ⚠️"
    details = [detail for _short, detail, _fine in sysinfo.speed_checks(False, 8)]
    assert "cryptg is missing" in details[0] and "uvloop is not running" in details[1] and "FAST_TRANSFER=0" in details[2]

    monkeypatch.setattr(sysinfo, "crypto_backend", lambda: "libssl")
    assert sysinfo.speed_summary(True, 16).startswith("libssl ✅") and "parallel ×16 ✅" in sysinfo.speed_summary(True, 16)


def test_the_speed_section_shows_the_stack_the_counts_and_the_last_failure(monkeypatch):
    from bot.fasttransfer import TransferStats

    monkeypatch.setattr(sysinfo, "crypto_backend", lambda: "cryptg")
    monkeypatch.setattr(sysinfo, "loop_backend", lambda: "uvloop")
    stats = TransferStats(fast_down=12, fast_up=9, small_up=3)
    text = sysinfo.format_speed(True, 8, stats)
    assert "⚡ <b>Speed</b>" in text and "Encryption: cryptg ✅" in text and "Event loop: uvloop ✅" in text
    assert "Parallel transfer: on, 8 connections ✅" in text
    assert "<b>Downloads:</b> 12 parallel · 0 standard" in text
    assert "<b>Uploads:</b> 9 parallel · 3 up to 10 MB · 0 standard" in text
    assert "failed" not in text and "⚠️" not in text

    stats.standard_down, stats.standard_up = 2, 1
    stats.fell_back("download", ValueError("<bad> & worse"))
    text = sysinfo.format_speed(False, 8, stats)
    assert "off (FAST_TRANSFER=0) ⚠️" in text and "2 standard" in text
    assert "failed 1×" in text and "<code>download: ValueError: &lt;bad&gt; &amp; worse</code>" in text  # HTML-safe


def test_format_status_adds_the_speed_section_only_when_given_one():
    plain = sysinfo.format_status(10, 1, "1.45.0")
    assert "Speed" not in plain and plain.endswith("Telethon 1.45.0")
    with_speed = sysinfo.format_status(10, 1, "1.45.0", "⚡ <b>Speed</b>\nline")
    assert with_speed.startswith(plain) and with_speed.endswith("\n\n⚡ <b>Speed</b>\nline")


def test_the_speed_summary_and_section_say_whether_streaming_is_on(monkeypatch):
    from bot.fasttransfer import TransferStats

    monkeypatch.setattr(sysinfo, "crypto_backend", lambda: "cryptg")
    monkeypatch.setattr(sysinfo, "loop_backend", lambda: "uvloop")
    assert sysinfo.speed_summary(True, 8, True) == "cryptg ✅ · uvloop ✅ · parallel ×8 ✅ · streaming ✅"
    assert sysinfo.speed_summary(True, 8, False).endswith("parallel ×8 ✅ · no streaming ⚠️")
    assert sysinfo.speed_summary(False, 8, True).endswith("parallel off ⚠️ · no streaming ⚠️")  # streaming needs the parallel transfer
    assert sysinfo.speed_summary(True, 8) == "cryptg ✅ · uvloop ✅ · parallel ×8 ✅"  # nothing added when it is not asked for

    stats = TransferStats(streamed=7)
    text = sysinfo.format_speed(True, 8, stats, True)
    assert "⬇⬆ Streaming rename: on: big files are uploaded while they download ✅" in text and "<b>Streamed renames:</b> 7" in text
    assert "Streaming failed" not in text and "⚠️" not in text
    assert "Streaming rename" not in sysinfo.format_speed(True, 8, stats) and "Streamed renames" not in sysinfo.format_speed(True, 8, stats)

    stats.stream_failed(ValueError("<x>"))
    text = sysinfo.format_speed(True, 8, stats, False)
    assert "off (STREAM_RENAME=0) ⚠️" in text
    assert "⚠️ Streaming failed 1× and those files were saved first and then uploaded. Last: <code>ValueError: &lt;x&gt;</code>" in text
    assert "The parallel transfer failed" not in text  # a streaming failure is not a parallel-transfer failure