Spaces:
Running
Running
| 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 | |
| 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" | |
| 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: <bad> & 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: <x></code>" in text | |
| assert "The parallel transfer failed" not in text # a streaming failure is not a parallel-transfer failure | |