pig.py/tests/unit/test_neopig.py

# 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.

"""
Tests for neopig module.

Tests the main NeoPig crawler class and related functionality.
"""

import pytest
from unittest.mock import Mock, AsyncMock, patch, MagicMock
import tempfile
import shutil
import os
import json
from pathlib import Path
from datetime import datetime, timezone

from neopig import (
    NeoPig,
    AppendOnlyStateLog,
    get_state_log_path,
    get_state_file_path,
    rotate_state_file,
    get_live_queue,
    emit_live_media,
    setup_logging,
    start_job_logging,
    stop_job_logging,
)


class TestAppendOnlyStateLog:
    """Test AppendOnlyStateLog class."""

    def setup_method(self):
        self.temp_dir = tempfile.mkdtemp()
        self.log_path = Path(self.temp_dir) / "test.log"

    def teardown_method(self):
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    def test_state_log_open_close(self):
        """Test opening and closing state log."""
        log = AppendOnlyStateLog(self.log_path)
        log.open()
        assert log._file is not None
        log.close()
        assert log._file is None

    def test_state_log_context_manager(self):
        """Test state log as context manager."""
        with AppendOnlyStateLog(self.log_path) as log:
            assert log._file is not None
        assert log._file is None

    def test_state_log_page(self):
        """Test recording page as seen."""
        with AppendOnlyStateLog(self.log_path) as log:
            log.page("https://example.com/page1")
            log.page("https://example.com/page2")

        content = self.log_path.read_text()
        assert "P https://example.com/page1" in content
        assert "P https://example.com/page2" in content

    def test_state_log_media(self):
        """Test recording media as downloaded."""
        with AppendOnlyStateLog(self.log_path) as log:
            log.media("abc123", "https://example.com/image.jpg")

        content = self.log_path.read_text()
        assert "M abc123 https://example.com/image.jpg" in content

    def test_state_log_screenshot(self):
        """Test recording screenshot as taken."""
        with AppendOnlyStateLog(self.log_path) as log:
            log.screenshot("https://example.com/page")

        content = self.log_path.read_text()
        assert "S https://example.com/page" in content

    def test_state_log_skip_domain(self):
        """Test recording domain to skip."""
        with AppendOnlyStateLog(self.log_path) as log:
            log.skip_domain("blocked.com")

        content = self.log_path.read_text()
        assert "D blocked.com" in content

    def test_state_log_stats(self):
        """Test recording stats checkpoint."""
        stats = {"pages_crawled": 100, "media_found": 500}
        with AppendOnlyStateLog(self.log_path) as log:
            log.stats(stats)

        content = self.log_path.read_text()
        assert "X stats" in content
        assert "100" in content

    def test_state_log_load_empty(self):
        """Test loading empty/nonexistent log."""
        log = AppendOnlyStateLog(self.log_path)
        result = log.load()

        assert result['seen_pages'] == set()
        assert result['seen_media'] == {}
        assert result['seen_screenshots'] == set()
        assert result['skip_domains'] == set()
        assert result['stats'] == {}

    def test_state_log_load_with_data(self):
        """Test loading log with data."""
        with AppendOnlyStateLog(self.log_path) as log:
            log.page("https://example.com/page1")
            log.media("abc123", "https://example.com/image.jpg")
            log.screenshot("https://example.com/")
            log.skip_domain("blocked.com")
            log.stats({"pages_crawled": 10})

        log = AppendOnlyStateLog(self.log_path)
        result = log.load()

        assert "https://example.com/page1" in result['seen_pages']
        assert result['seen_media'].get("https://example.com/image.jpg") == "abc123"
        assert "https://example.com/" in result['seen_screenshots']
        assert "blocked.com" in result['skip_domains']
        assert result['stats'].get('pages_crawled') == 10


class TestStateFilePaths:
    """Test state file path generation."""

    def test_get_state_log_path(self):
        """Test state log path generation."""
        path = get_state_log_path("example.com")
        assert path == Path("data/example-com.log")

    def test_get_state_log_path_with_subdomain(self):
        """Test state log path with subdomain."""
        path = get_state_log_path("sub.example.com")
        assert "sub" in str(path)
        assert path.suffix == ".log"

    def test_get_state_file_path(self):
        """Test state file path generation."""
        path = get_state_file_path("example.com")
        assert path == Path("data/example-com.state")

    def test_get_state_file_path_special_chars(self):
        """Test state file path with special characters."""
        path = get_state_file_path("example.com:8080")
        assert "state" in str(path)
        # The path should be a valid state file path
        assert path.suffix == ".state"


class TestRotateStateFile:
    """Test state file rotation."""

    def setup_method(self):
        self.temp_dir = tempfile.mkdtemp()

    def teardown_method(self):
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    def test_rotate_nonexistent(self):
        """Test rotating nonexistent file returns None."""
        path = Path(self.temp_dir) / "nonexistent.state"
        result = rotate_state_file(path)
        assert result is None

    def test_rotate_existing(self):
        """Test rotating existing file."""
        path = Path(self.temp_dir) / "test.state"
        path.write_text('{"key": "value"}')

        rotated = rotate_state_file(path)

        assert rotated is not None
        assert rotated.exists()
        assert not path.exists()
        assert ".1" in str(rotated)

    def test_rotate_multiple(self):
        """Test multiple rotations increment number."""
        path = Path(self.temp_dir) / "test.state"

        # First rotation
        path.write_text('{"v": 1}')
        rotated1 = rotate_state_file(path)
        assert ".1" in str(rotated1)

        # Second rotation
        path.write_text('{"v": 2}')
        rotated2 = rotate_state_file(path)
        assert ".2" in str(rotated2)

    def test_rotate_with_preserve_keys(self):
        """Test rotation with key preservation."""
        path = Path(self.temp_dir) / "test.state"
        path.write_text('{"keep": "value", "discard": "other"}')

        rotated = rotate_state_file(path, preserve_keys=["keep"])

        assert rotated.exists()
        assert path.exists()  # New file with preserved keys
        new_content = json.loads(path.read_text())
        assert new_content.get("keep") == "value"
        assert "discard" not in new_content


class TestLiveQueue:
    """Test live media queue functions."""

    def test_get_live_queue(self):
        """Test getting live queue creates queue."""
        queue = get_live_queue()
        assert queue is not None
        # Should return same queue on second call
        queue2 = get_live_queue()
        assert queue is queue2

    def test_emit_live_media(self):
        """Test emitting media to queue."""
        queue = get_live_queue()
        # Empty queue first
        while not queue.empty():
            try:
                queue.get_nowait()
            except:
                break

        media_info = {"md5_hash": "abc123", "media_type": "image"}
        emit_live_media(media_info)

        # Should be able to get the item
        assert not queue.empty()


class TestNeoPigInitialization:
    """Test NeoPig class initialization."""

    def setup_method(self):
        self.temp_dir = tempfile.mkdtemp()
        self.db_path = os.path.join(self.temp_dir, "test.db")
        self.vault_path = os.path.join(self.temp_dir, "vault")

    def teardown_method(self):
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    def test_neopig_creation(self):
        """Test creating NeoPig instance."""
        pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
        )

        assert pig.db is not None
        assert pig.vault is not None
        assert pig.fetcher is not None
        assert pig.screenshot is not None

    def test_neopig_fast_mode(self):
        """Test NeoPig in fast mode."""
        pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
            fast_mode=True,
        )

        assert pig.fast_mode is True
        assert pig.fetcher.default_crawl_delay == 0.0

    def test_neopig_stats_initialized(self):
        """Test NeoPig stats are initialized."""
        pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
        )

        assert pig.stats['pages_crawled'] == 0
        assert pig.stats['media_found'] == 0
        assert pig.stats['errors'] == 0

    def test_neopig_state_tracking(self):
        """Test NeoPig state tracking structures initialized."""
        pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
        )

        assert pig.seen_media == {}
        assert pig.seen_screenshots == set()
        assert pig.seen_pages == set()


class TestNeoPigStateMethods:
    """Test NeoPig state management methods."""

    def setup_method(self):
        self.temp_dir = tempfile.mkdtemp()
        self.db_path = os.path.join(self.temp_dir, "test.db")
        self.vault_path = os.path.join(self.temp_dir, "vault")
        self.pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
        )
        # Override state dir for tests
        self.pig._state_dir = Path(self.temp_dir)

    def teardown_method(self):
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    def test_get_state_file(self):
        """Test getting state file path for URL."""
        path = self.pig._get_state_file("https://example.com/page")
        assert "example" in str(path)
        assert path.suffix == ".state"

    def test_get_state_log(self):
        """Test getting state log for URL."""
        log = self.pig._get_state_log("https://example.com/page")
        # Check by class name since module identity differs between package and module
        assert log.__class__.__name__ == 'AppendOnlyStateLog'

    def test_get_hydra_state_file(self):
        """Test getting hydra state file path."""
        path = self.pig._get_hydra_state_file("example.com")
        assert "hydra" in str(path)
        assert ".json" in str(path)

    def test_load_hydra_state_empty(self):
        """Test loading empty hydra state."""
        state = self.pig._load_hydra_state("example.com")
        assert state['seen_urls'] == {}
        assert state['feeds'] == {}

    def test_save_and_load_hydra_state(self):
        """Test saving and loading hydra state."""
        domain = "example.com"
        state = {
            'seen_urls': {'https://example.com/article': {}},
            'feeds': {'https://example.com/rss': {}},
        }
        self.pig._save_hydra_state(domain, state)

        loaded = self.pig._load_hydra_state(domain)
        assert 'https://example.com/article' in loaded['seen_urls']
        assert 'https://example.com/rss' in loaded['feeds']

    def test_get_known_feeds(self):
        """Test getting known feeds for domain."""
        domain = "example.com"
        state = {
            'seen_urls': {},
            'feeds': {
                'https://example.com/rss': {},
                'https://example.com/atom': {},
            },
        }
        self.pig._save_hydra_state(domain, state)

        feeds = self.pig._get_known_feeds(domain)
        assert len(feeds) == 2
        assert 'https://example.com/rss' in feeds

    def test_add_hydra_feeds(self):
        """Test adding new feeds to hydra state."""
        domain = "example.com"
        self.pig._add_hydra_feeds(domain, ['https://example.com/feed1'])

        state = self.pig._load_hydra_state(domain)
        assert 'https://example.com/feed1' in state['feeds']


class TestNeoPigAsync:
    """Test NeoPig async methods."""

    def setup_method(self):
        self.temp_dir = tempfile.mkdtemp()
        self.db_path = os.path.join(self.temp_dir, "test.db")
        self.vault_path = os.path.join(self.temp_dir, "vault")

    def teardown_method(self):
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    @pytest.mark.asyncio
    async def test_neopig_init_async(self):
        """Test NeoPig async initialization."""
        pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
        )
        await pig.init()

        assert pig.db._initialized is True
        assert pig.vault._initialized is True

        # Cleanup
        await pig.db.close()

    @pytest.mark.asyncio
    async def test_neopig_db_close(self):
        """Test NeoPig database cleanup."""
        pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
        )
        await pig.init()
        await pig.db.close()

        # Should not raise


class TestJobLogging:
    """Test job logging functions."""

    def setup_method(self):
        self.temp_dir = tempfile.mkdtemp()

    def teardown_method(self):
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    def test_start_job_logging(self):
        """Test starting job logging."""
        # This modifies global state, so we just verify it doesn't crash
        # LOGS_PATH is now in neopig.logging module
        # The actual logging is tested in test_neopig_logging.py
        pass


class TestSetupLogging:
    """Test logging setup."""

    def test_setup_logging(self):
        """Test setting up logging doesn't crash."""
        import logging
        setup_logging(level=logging.DEBUG)
        # Verify root logger has handlers
        assert len(logging.getLogger().handlers) > 0


class TestNeoPigStats:
    """Test NeoPig statistics tracking."""

    def setup_method(self):
        self.temp_dir = tempfile.mkdtemp()
        self.db_path = os.path.join(self.temp_dir, "test.db")
        self.vault_path = os.path.join(self.temp_dir, "vault")
        self.pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
        )

    def teardown_method(self):
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    def test_stats_structure(self):
        """Test stats dictionary structure."""
        required_keys = [
            'pages_crawled',
            'pages_pending',
            'media_found',
            'media_downloaded',
            'duplicates_skipped',
            'screenshots_taken',
            'errors',
            'bytes_downloaded',
            'bytes_stored',
        ]
        for key in required_keys:
            assert key in self.pig.stats

    def test_stats_initial_values(self):
        """Test stats start at zero."""
        for key, value in self.pig.stats.items():
            if isinstance(value, int):
                assert value == 0


class TestNeoPigVaultManager:
    """Test NeoPig vault manager integration."""

    def setup_method(self):
        self.temp_dir = tempfile.mkdtemp()
        self.db_path = os.path.join(self.temp_dir, "test.db")
        self.vault_path = os.path.join(self.temp_dir, "vault")
        self.pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
        )

    def teardown_method(self):
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    def test_domain_vaults_initialized(self):
        """Test domain vaults are initialized."""
        assert self.pig.domain_vaults is not None

    def test_vault_paths(self):
        """Test vault paths are set correctly."""
        assert self.pig.vault_path == self.vault_path


class TestTqdmLoggingHandler:
    """Test TqdmLoggingHandler."""

    def test_tqdm_handler_exists(self):
        """Test TqdmLoggingHandler class exists."""
        from neopig import TqdmLoggingHandler
        handler = TqdmLoggingHandler()
        assert handler is not None

    def test_tqdm_handler_emit(self):
        """Test handler emit doesn't crash."""
        from neopig import TqdmLoggingHandler
        import logging

        handler = TqdmLoggingHandler()
        handler.setFormatter(logging.Formatter('%(message)s'))

        record = logging.LogRecord(
            name='test',
            level=logging.INFO,
            pathname='test.py',
            lineno=1,
            msg='Test message',
            args=(),
            exc_info=None
        )

        # Should not raise
        handler.emit(record)


class TestNeoPigCrawlMethods:
    """Test NeoPig crawl-related methods."""

    def setup_method(self):
        self.temp_dir = tempfile.mkdtemp()
        self.db_path = os.path.join(self.temp_dir, "test.db")
        self.vault_path = os.path.join(self.temp_dir, "vault")
        self.pig = NeoPig(
            db_path=self.db_path,
            vault_path=self.vault_path,
        )
        self.pig._state_dir = Path(self.temp_dir)

    def teardown_method(self):
        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    def test_save_state_interval(self):
        """Test state save interval."""
        # State is saved every N items
        assert self.pig._state_save_interval == 10
        assert self.pig._items_since_save == 0

    @pytest.mark.asyncio
    async def test_neopig_crawl_returns_stats(self):
        """Test that crawl stats structure exists."""
        await self.pig.init()

        # We can't fully test crawl without extensive mocking,
        # but we can verify the structure
        assert 'pages_crawled' in self.pig.stats
        assert 'media_found' in self.pig.stats
        assert 'errors' in self.pig.stats

        await self.pig.db.close()


if __name__ == '__main__':
    pytest.main([__file__, '-v'])