# retoor from datetime import datetime, timedelta, timezone import pytest from devplacepy.database import ( CONSENT_KINDS, SOFT_DELETE_TABLES, UNREPORTABLE_TABLES, consent_granted, get_setting, get_table, ) from devplacepy.database import nodes as store from devplacepy.services.moderation.deletion import OWNED_TABLES from devplacepy.services.nodes.ledger import UsageLedger from devplacepy.services.nodes.spec import SPEC from devplacepy.utils import generate_uid LIMITS = SPEC.limits SOFT_TABLES = ("node_devices", "node_settings", "node_usage_daily", "node_donors", "node_flags") @pytest.fixture def owner(local_db): return generate_uid() @pytest.fixture def device(owner): row, _ = store.create_device(owner, "laptop", "native") return row def test_node_tables_are_registered_everywhere_they_must_be(): for table in SOFT_TABLES: assert table in SOFT_DELETE_TABLES assert table in UNREPORTABLE_TABLES for table in ("node_devices", "node_settings", "node_donors", "node_usage_daily"): assert (table, "user_uid") in OWNED_TABLES def test_bandwidth_sharing_consent_exists_and_is_never_seeded(owner): assert "bandwidth_sharing" in CONSENT_KINDS assert not consent_granted("user", owner, "bandwidth_sharing") assert get_setting("nodes_enabled", "0") == "0" def test_secret_is_shown_once_and_stored_hashed(owner): row, secret = store.create_device(owner, " desk ", "extension") assert secret.startswith(store.SECRET_PREFIX) assert row["label"] == "desk" assert secret not in str(row) assert store.find_device_by_secret(secret)["uid"] == row["uid"] assert store.find_device_by_secret(secret + "x") is None assert store.find_device_by_secret("") is None def test_rotate_invalidates_the_old_secret_and_checks_the_owner(owner): row, old = store.create_device(owner, "a", "native") assert store.rotate_device_secret(row["uid"], generate_uid()) is None new = store.rotate_device_secret(row["uid"], owner) assert new and new != old assert store.find_device_by_secret(old) is None assert store.find_device_by_secret(new)["uid"] == row["uid"] def test_device_cap_per_user_is_enforced(owner): for _ in range(LIMITS.max_devices_per_user): store.create_device(owner, "x", "native") with pytest.raises(store.NodeLimitError): store.create_device(owner, "x", "native") assert store.count_devices(owner) == LIMITS.max_devices_per_user def test_unknown_device_kind_is_rejected(owner): with pytest.raises(store.NodeLimitError): store.create_device(owner, "x", "browser") def test_revoke_is_a_soft_delete_and_frees_the_slot(owner): row, secret = store.create_device(owner, "a", "native") assert store.revoke_device(row["uid"], owner) == 1 assert store.get_device(row["uid"]) is None assert store.find_device_by_secret(secret) is None assert store.count_devices(owner) == 0 assert get_table("node_devices").find_one(uid=row["uid"])["deleted_by"] == owner def test_suspension_expires_on_its_own_and_revoked_never_suspends(owner): row, _ = store.create_device(owner, "a", "native") assert store.suspend_device(row["uid"], "integrity", 30) suspended = store.get_device(row["uid"]) assert store.effective_status(suspended) == "suspended" later = datetime.now(timezone.utc) + timedelta(minutes=31) assert store.effective_status(suspended, later) == "active" assert store.unsuspend_device(row["uid"]) assert store.get_device(row["uid"])["status"] == "active" store.revoke_device(row["uid"], owner) assert not store.suspend_device(row["uid"], "x", 5) def test_settings_default_to_opted_out(owner): settings = store.get_node_settings(owner) assert settings["enabled"] is False assert settings["bandwidth_mb_day"] == LIMITS.default_donor_mb_per_day @pytest.mark.parametrize("value", [-5, 0, 1, 500, 10**9, "x", None, 3.7]) def test_settings_are_clamped_to_the_declared_ceilings(owner, value): saved = store.save_node_settings( owner, {"bandwidth_mb_day": value, "max_concurrent": value} ) assert 1 <= saved["bandwidth_mb_day"] <= LIMITS.max_donor_mb_per_day assert 1 <= saved["max_concurrent"] <= LIMITS.max_concurrent_per_node assert store.get_node_settings(owner) == saved def test_settings_reject_bad_hours_and_unknown_platforms(owner): saved = store.save_node_settings( owner, {"enabled": True, "active_from": "25:00", "active_to": "07:30", "allowed_platforms": ["reddit", "evil"]}, ) assert saved["active_from"] == "" assert saved["active_to"] == "07:30" assert saved["allowed_platforms"] == ["reddit"] again = store.save_node_settings(owner, {"enabled": False}) assert again["active_to"] == "07:30" assert again["enabled"] is False def test_daily_cap_blocks_the_job_that_would_cross_it(owner, device): cap = 10_000 results = [ store.record_job(device["uid"], owner, ok=True, bytes_up=100, bytes_down=900, cap_bytes=cap) for _ in range(14) ] assert results.count(True) == 10 today = store.usage_today(device["uid"]) assert today["requests_ok"] == 10 assert today["bytes_up"] + today["bytes_down"] == cap donor = store.get_donor(owner) assert donor["requests_ok"] == 10 and donor["bytes_total"] == cap def test_failed_jobs_count_but_move_no_bytes_or_lifetime_totals(owner, device): store.record_job(device["uid"], owner, ok=False, bytes_down=999) today = store.usage_today(device["uid"]) assert today["requests_failed"] == 1 assert today["bytes_down"] == 0 assert store.get_donor(owner)["requests_ok"] == 0 def test_lifetime_totals_survive_pruning_the_daily_buckets(owner, device): store.record_job(device["uid"], owner, ok=True, bytes_down=10, cap_bytes=1000) far = datetime.now(timezone.utc) + timedelta(days=LIMITS.usage_daily_retention_days + 2) assert store.prune_usage_daily(dry_run=True, moment=far) >= 1 assert store.usage_today(device["uid"])["requests_ok"] == 1 assert store.prune_usage_daily(moment=far) >= 1 assert store.usage_today(device["uid"])["requests_ok"] == 0 assert store.get_donor(owner)["requests_ok"] == 1 def test_recent_buckets_are_not_pruned(owner, device): store.record_job(device["uid"], owner, ok=True, cap_bytes=1000) assert store.prune_usage_daily() == 0 def test_xp_is_capped_per_day_and_resets_the_next_day(owner, device): store.record_job(device["uid"], owner, ok=True, cap_bytes=1000) granted = sum(store.claim_xp(owner, 7) for _ in range(30)) assert granted == LIMITS.xp_daily_cap // 7 tomorrow = datetime.now(timezone.utc) + timedelta(days=1) assert store.claim_xp(owner, 7, moment=tomorrow) assert not store.claim_xp(owner, 0) assert not store.claim_xp(generate_uid(), 7) def test_integrity_failures_are_counted(owner, device): store.record_integrity_failure(device["uid"], owner) store.record_integrity_failure(device["uid"], owner) assert store.usage_today(device["uid"])["integrity_failures"] == 2 def test_ledger_summary_reports_lifetime_month_and_today(owner, device): ledger = UsageLedger() assert ledger.record_success(device["uid"], owner, bytes_up=10, bytes_down=90) ledger.record_failure(device["uid"], owner) summary = ledger.summary(owner) assert summary.lifetime_requests == 1 assert summary.lifetime_bytes == 100 assert summary.month_requests == 1 assert summary.today_bytes == 100 assert summary.first_donated_at def test_community_stats_are_aggregate_only(owner, device): store.record_job(device["uid"], owner, ok=True, bytes_down=50, cap_bytes=1000) stats = store.community_stats() assert set(stats) == {"requests", "bytes", "donors"} assert stats["requests"] >= 1 and stats["donors"] >= 1 def test_flags_deduplicate_while_open_and_resolve_once(owner): flag = store.raise_flag("integrity", device_uid="d1", user_uid=owner, detail="x") assert store.raise_flag("integrity", device_uid="d1")["uid"] == flag["uid"] assert store.resolve_flag(flag["uid"], "admin") assert not store.resolve_flag(flag["uid"], "admin") assert flag["uid"] not in [item["uid"] for item in store.list_flags()] def test_event_log_is_bounded_by_age(owner): store.add_event("connect", user_uid=owner, detail="x" * 1000) row = get_table("node_events").find_one(user_uid=owner) assert len(row["detail"]) == store.EVENT_DETAIL_MAX far = datetime.now(timezone.utc) + timedelta(days=LIMITS.event_retention_days + 1) assert store.prune_events(dry_run=True, moment=far) >= 1 assert store.prune_events(moment=far) >= 1 def test_ip_is_only_ever_stored_hashed(): digest = store.hash_ip("203.0.113.9") assert "203" not in digest and len(digest) == 32 assert store.hash_ip("203.0.113.9") == digest assert store.hash_ip("") == ""