125 lines
5.0 KiB
Python
125 lines
5.0 KiB
Python
import asyncio
|
|
import sys
|
|
|
|
sys.path.insert(0, "/app")
|
|
|
|
from db.database import init_db, close_db_pool
|
|
from db.repository import Repository
|
|
from db.models import ChannelCategory, SourceChannel, TargetChannel
|
|
from services.admin_bot import AdminBotService
|
|
|
|
SRC_1 = -1009999000031
|
|
SRC_2 = -1009999000032
|
|
TRG_1 = -1009999000033
|
|
TRG_2 = -1009999000034
|
|
|
|
|
|
async def _cleanup(repo):
|
|
pool = await repo._get_pool()
|
|
async with pool.acquire() as conn:
|
|
for cid in (SRC_1, SRC_2):
|
|
await conn.execute("DELETE FROM sources WHERE channel_id = $1;", cid)
|
|
for cid in (TRG_1, TRG_2):
|
|
await conn.execute("DELETE FROM targets WHERE channel_id = $1;", cid)
|
|
await conn.execute("DELETE FROM channel_categories WHERE name LIKE 'Test Cat%';")
|
|
|
|
|
|
async def run_tests():
|
|
await init_db()
|
|
repo = Repository()
|
|
await _cleanup(repo)
|
|
|
|
# 1. Create categories
|
|
cat1_id = await repo.create_category("Test Cat Tech", "both", "Technology news")
|
|
cat2_id = await repo.create_category("Test Cat Crypto", "both", "Crypto & Finance")
|
|
assert cat1_id > 0
|
|
assert cat2_id > 0
|
|
|
|
cats = await repo.get_categories()
|
|
cat_names = [c.name for c in cats]
|
|
assert "Test Cat Tech" in cat_names
|
|
assert "Test Cat Crypto" in cat_names
|
|
|
|
# 2. Add Sources & Targets
|
|
s1_id = await repo.add_source(SRC_1, "Tech Source 1", "tech_src1")
|
|
s2_id = await repo.add_source(SRC_2, "Crypto Source 2", "crypto_src2")
|
|
t1_id = await repo.add_target(TRG_1, "Tech Target 1", "tech_trg1")
|
|
t2_id = await repo.add_target(TRG_2, "Crypto Target 2", "crypto_trg2")
|
|
|
|
# Initial state: no category
|
|
assert (await repo.get_source_by_id(s1_id)).category_id is None
|
|
assert (await repo.get_target_by_id(t1_id)).category_id is None
|
|
|
|
# 3. Assign categories
|
|
await repo.set_source_category(s1_id, cat1_id)
|
|
await repo.set_source_category(s2_id, cat2_id)
|
|
await repo.set_target_category(t1_id, cat1_id)
|
|
await repo.set_target_category(t2_id, cat2_id)
|
|
|
|
# Verify assignments
|
|
assert (await repo.get_source_by_id(s1_id)).category_id == cat1_id
|
|
assert (await repo.get_source_by_id(s2_id)).category_id == cat2_id
|
|
assert (await repo.get_target_by_id(t1_id)).category_id == cat1_id
|
|
assert (await repo.get_target_by_id(t2_id)).category_id == cat2_id
|
|
|
|
# 4. Filter channels by category
|
|
tech_sources = await repo.get_sources_by_category(cat1_id)
|
|
tech_targets = await repo.get_targets_by_category(cat1_id)
|
|
assert len(tech_sources) == 1 and tech_sources[0].id == s1_id
|
|
assert len(tech_targets) == 1 and tech_targets[0].id == t1_id
|
|
|
|
# 5. Test AdminBot category-based rendering
|
|
bot = AdminBotService(repo=repo, ai_processor=None, queue=None, review_channel_id=0, admin_user_ids=[1])
|
|
|
|
src_text, src_buttons = await bot._render_source_list()
|
|
assert "دستهبندیهای کانالهای مبدا" in src_text
|
|
btn_data = [btn.data.decode("utf-8") for row in src_buttons for btn in row]
|
|
assert f"src_cat_view:{cat1_id}" in btn_data
|
|
assert f"src_cat_view:{cat2_id}" in btn_data
|
|
|
|
# Test clicking a category for sources
|
|
cat_src_text, cat_src_buttons = await bot._render_source_channels_in_category(cat1_id)
|
|
assert "کانالهای مبدا در دسته" in cat_src_text
|
|
assert "Tech Cat" in cat_src_text or "Test Cat Tech" in cat_src_text
|
|
cat_btn_data = [btn.data.decode("utf-8") for row in cat_src_buttons for btn in row]
|
|
assert f"src_view:{s1_id}" in cat_btn_data
|
|
|
|
# Test target category navigation
|
|
trg_text, trg_buttons = await bot._render_target_list()
|
|
assert "دستهبندیهای کانالهای مقصد" in trg_text
|
|
trg_btn_data = [btn.data.decode("utf-8") for row in trg_buttons for btn in row]
|
|
assert f"trg_cat_view:{cat1_id}" in trg_btn_data
|
|
|
|
cat_trg_text, cat_trg_buttons = await bot._render_target_channels_in_category(cat1_id)
|
|
assert "کانالهای مقصد در دسته" in cat_trg_text
|
|
cat_trg_btn_data = [btn.data.decode("utf-8") for row in cat_trg_buttons for btn in row]
|
|
assert f"trg_view:{t1_id}" in cat_trg_btn_data
|
|
|
|
# 6. Counts
|
|
counts = await repo.get_category_channel_counts(cat1_id)
|
|
assert counts["sources"] == 1
|
|
assert counts["targets"] == 1
|
|
|
|
# 7. Rename category
|
|
await repo.update_category(cat1_id, name="Test Cat Tech Updated")
|
|
updated_cat = await repo.get_category_by_id(cat1_id)
|
|
assert updated_cat.name == "Test Cat Tech Updated"
|
|
|
|
# 8. Unassign category
|
|
await repo.set_source_category(s1_id, None)
|
|
assert (await repo.get_source_by_id(s1_id)).category_id is None
|
|
|
|
# 9. Delete category (must nullify channel references safely)
|
|
await repo.delete_category(cat2_id)
|
|
assert await repo.get_category_by_id(cat2_id) is None
|
|
assert (await repo.get_source_by_id(s2_id)).category_id is None
|
|
assert (await repo.get_target_by_id(t2_id)).category_id is None
|
|
|
|
await _cleanup(repo)
|
|
await close_db_pool()
|
|
print("All category database, repository, and UI rendering tests passed successfully!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_tests())
|