anti
f3bb0b31ae
New GET /attackers/{uuid}/commands?limit=&offset=&service= endpoint
serves commands with server-side pagination and optional service filter.
AttackerDetail frontend fetches commands from this endpoint with
page controls. Service badge filter now drives both the API query
and the local fingerprint filter.
64 lines
3.0 KiB
Python
64 lines
3.0 KiB
Python
"""
|
|
Mock test for BaseRepository to ensure coverage of abstract pass lines.
|
|
"""
|
|
|
|
import pytest
|
|
from decnet.web.db.repository import BaseRepository
|
|
|
|
class DummyRepo(BaseRepository):
|
|
async def initialize(self) -> None: await super().initialize()
|
|
async def add_log(self, data): await super().add_log(data)
|
|
async def get_logs(self, **kw): await super().get_logs(**kw)
|
|
async def get_total_logs(self, **kw): await super().get_total_logs(**kw)
|
|
async def get_stats_summary(self): await super().get_stats_summary()
|
|
async def get_deckies(self): await super().get_deckies()
|
|
async def get_user_by_username(self, u): await super().get_user_by_username(u)
|
|
async def get_user_by_uuid(self, u): await super().get_user_by_uuid(u)
|
|
async def create_user(self, d): await super().create_user(d)
|
|
async def update_user_password(self, *a, **kw): await super().update_user_password(*a, **kw)
|
|
async def add_bounty(self, d): await super().add_bounty(d)
|
|
async def get_bounties(self, **kw): await super().get_bounties(**kw)
|
|
async def get_total_bounties(self, **kw): await super().get_total_bounties(**kw)
|
|
async def get_state(self, k): await super().get_state(k)
|
|
async def set_state(self, k, v): await super().set_state(k, v)
|
|
async def get_all_logs_raw(self): await super().get_all_logs_raw()
|
|
async def get_max_log_id(self): await super().get_max_log_id()
|
|
async def get_logs_after_id(self, last_id, limit=500): await super().get_logs_after_id(last_id, limit)
|
|
async def get_all_bounties_by_ip(self): await super().get_all_bounties_by_ip()
|
|
async def get_bounties_for_ips(self, ips): await super().get_bounties_for_ips(ips)
|
|
async def upsert_attacker(self, d): await super().upsert_attacker(d)
|
|
async def get_attacker_by_uuid(self, u): await super().get_attacker_by_uuid(u)
|
|
async def get_attackers(self, **kw): await super().get_attackers(**kw)
|
|
async def get_total_attackers(self, **kw): await super().get_total_attackers(**kw)
|
|
async def get_attacker_commands(self, **kw): await super().get_attacker_commands(**kw)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_base_repo_coverage():
|
|
dr = DummyRepo()
|
|
# Call all to hit 'pass' statements
|
|
await dr.initialize()
|
|
await dr.add_log({})
|
|
await dr.get_logs()
|
|
await dr.get_total_logs()
|
|
await dr.get_stats_summary()
|
|
await dr.get_deckies()
|
|
await dr.get_user_by_username("a")
|
|
await dr.get_user_by_uuid("a")
|
|
await dr.create_user({})
|
|
await dr.update_user_password("a", "b")
|
|
await dr.add_bounty({})
|
|
await dr.get_bounties()
|
|
await dr.get_total_bounties()
|
|
await dr.get_state("k")
|
|
await dr.set_state("k", "v")
|
|
await dr.get_all_logs_raw()
|
|
await dr.get_max_log_id()
|
|
await dr.get_logs_after_id(0)
|
|
await dr.get_all_bounties_by_ip()
|
|
await dr.get_bounties_for_ips({"1.1.1.1"})
|
|
await dr.upsert_attacker({})
|
|
await dr.get_attacker_by_uuid("a")
|
|
await dr.get_attackers()
|
|
await dr.get_total_attackers()
|
|
await dr.get_attacker_commands(uuid="a")
|