149 lines
4.6 KiB
Python
149 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
# This is free software for the public good of a permacomputer hosted at
|
|
# permacomputer.com, an always-on computer by the people, for the people.
|
|
# One which is durable, easy to repair, & distributed like tap water
|
|
# for machine learning intelligence.
|
|
#
|
|
# The permacomputer is community-owned infrastructure optimized around
|
|
# four values:
|
|
#
|
|
# TRUTH First principles, math & science, open source code freely distributed
|
|
# FREEDOM Voluntary partnerships, freedom from tyranny & corporate control
|
|
# HARMONY Minimal waste, self-renewing systems with diverse thriving connections
|
|
# LOVE Be yourself without hurting others, cooperation through natural law
|
|
#
|
|
# This software contributes to that vision by archiving the web, preserving digital knowledge before it disappears.
|
|
# Code is seeds to sprout on any abandoned technology.
|
|
|
|
"""
|
|
Functional test for neopig crawler.
|
|
|
|
Tests crawling upload.unturf.com for images.
|
|
|
|
These tests hit live network and are marked with @pytest.mark.network.
|
|
To skip in CI: pytest -m "not network"
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# Add current dir to path
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from neopig import NeoPig
|
|
from neopig.async_web_fetcher import CrawlMode
|
|
|
|
|
|
@pytest.mark.network
|
|
async def test_crawl_upload_unturf(tmp_path):
|
|
"""Test crawling upload.unturf.com for images."""
|
|
print("=" * 60)
|
|
print("neopig functional test")
|
|
print("Target: https://upload.unturf.com")
|
|
print("=" * 60)
|
|
|
|
# Use temp directory for test database and vault
|
|
db_path = tmp_path / "test_neopig.db"
|
|
vault_path = tmp_path / "test_vault"
|
|
|
|
pig = NeoPig(
|
|
db_path=str(db_path),
|
|
vault_path=str(vault_path)
|
|
)
|
|
await pig.init()
|
|
|
|
print("\n[1/3] Starting crawl...")
|
|
stats = await pig.crawl(
|
|
target_uri="https://upload.unturf.com",
|
|
keywords=["unturf", "upload"],
|
|
mode=CrawlMode.IMAGES,
|
|
depth=-1, # Unlimited depth
|
|
max_pages=100, # More pages
|
|
download_media=True,
|
|
)
|
|
|
|
print("\n[2/3] Crawl complete!")
|
|
print(f" Pages crawled: {stats['pages_crawled']}")
|
|
print(f" Media found: {stats['media_found']}")
|
|
print(f" Media downloaded: {stats['media_downloaded']}")
|
|
print(f" Duplicates skipped: {stats['duplicates_skipped']}")
|
|
print(f" Errors: {stats['errors']}")
|
|
|
|
# Check vault (filevault)
|
|
print("\n[3/3] Checking filevault...")
|
|
vault_stats = await pig.vault.stats()
|
|
print(f" Vault path: {vault_stats['path']}")
|
|
print(f" Vault depth: {vault_stats['depth']}")
|
|
print(f" Files stored: {vault_stats['count']}")
|
|
if 'total_size' in vault_stats:
|
|
size_mb = vault_stats['total_size'] / (1024 * 1024)
|
|
print(f" Total size: {size_mb:.2f} MB")
|
|
|
|
# Check database
|
|
from neopig.database import Database
|
|
db = Database(str(db_path))
|
|
await db.init()
|
|
db_stats = await db.get_stats()
|
|
print(f"\n Database stats:")
|
|
print(f" Total media: {db_stats['total_media']}")
|
|
print(f" By type: {db_stats['by_type']}")
|
|
print(f" Total sources: {db_stats['total_sources']}")
|
|
|
|
print("\n" + "=" * 60)
|
|
if stats['media_downloaded'] > 0:
|
|
print("TEST PASSED - Successfully crawled and stored images")
|
|
elif stats['media_found'] > 0:
|
|
print("TEST PARTIAL - Found images but none downloaded (may be dupes)")
|
|
else:
|
|
print("TEST WARNING - No images found on target")
|
|
print("=" * 60)
|
|
|
|
return stats
|
|
|
|
|
|
@pytest.mark.network
|
|
async def test_crawl_modes(tmp_path):
|
|
"""Test different crawl modes."""
|
|
print("\n" + "=" * 60)
|
|
print("Testing crawl modes...")
|
|
print("=" * 60)
|
|
|
|
# Use temp directory for test database and vault
|
|
db_path = tmp_path / "test_neopig.db"
|
|
vault_path = tmp_path / "test_vault"
|
|
|
|
pig = NeoPig(
|
|
db_path=str(db_path),
|
|
vault_path=str(vault_path)
|
|
)
|
|
await pig.init()
|
|
|
|
# Test media mode (images + videos + audio)
|
|
print("\n[MEDIA mode] Crawling for all media...")
|
|
stats = await pig.crawl(
|
|
target_uri="https://upload.unturf.com",
|
|
keywords=["test"],
|
|
mode=CrawlMode.MEDIA,
|
|
depth=1,
|
|
max_pages=5,
|
|
download_media=False, # Just index, don't download
|
|
)
|
|
print(f" Found: {stats['media_found']} media items (not downloaded)")
|
|
|
|
return stats
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import tempfile
|
|
print("neopig functional test suite\n")
|
|
|
|
# Run tests with temp directory
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
asyncio.run(test_crawl_upload_unturf(tmp_path))
|
|
asyncio.run(test_crawl_modes(tmp_path))
|
|
|
|
print("\nDone!")
|