#!/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!")