← back to Handbag Auth Nextjs

scripts/tests/test_scraper.py

540 lines

#!/usr/bin/env python3
"""
Unit and Integration Tests for Auction Scraper
"""

import sys
import os
import sqlite3
import pytest
import tempfile
import shutil
from datetime import datetime
from unittest.mock import Mock, patch, MagicMock
import requests_mock

# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# Import the scraper module
import importlib.util
spec = importlib.util.spec_from_file_location("scraper", os.path.join(os.path.dirname(os.path.dirname(__file__)), "scrape-auction-data.py"))
scraper = importlib.util.module_from_spec(spec)


class TestDatabaseInitialization:
    """Test database initialization and schema"""

    def test_database_creation(self, tmp_path):
        """Test that database file is created"""
        db_path = tmp_path / "test.db"
        conn = sqlite3.connect(str(db_path))
        cursor = conn.cursor()

        cursor.execute("""
            CREATE TABLE IF NOT EXISTS auctions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                auction_id TEXT UNIQUE,
                title TEXT,
                auction_house TEXT,
                current_price REAL,
                estimate_low REAL,
                estimate_high REAL,
                end_date TEXT,
                lot_number TEXT,
                url TEXT,
                search_term TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.commit()
        conn.close()

        assert os.path.exists(db_path)

    def test_schema_has_required_columns(self, tmp_path):
        """Test that schema has all required columns"""
        db_path = tmp_path / "test.db"
        conn = sqlite3.connect(str(db_path))
        cursor = conn.cursor()

        cursor.execute("""
            CREATE TABLE IF NOT EXISTS auctions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                auction_id TEXT UNIQUE,
                title TEXT,
                auction_house TEXT,
                current_price REAL,
                estimate_low REAL,
                estimate_high REAL,
                end_date TEXT,
                lot_number TEXT,
                url TEXT,
                search_term TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        cursor.execute("PRAGMA table_info(auctions)")
        columns = [row[1] for row in cursor.fetchall()]

        required_columns = [
            'id', 'auction_id', 'title', 'auction_house',
            'current_price', 'estimate_low', 'estimate_high',
            'end_date', 'lot_number', 'url', 'search_term', 'created_at'
        ]

        for col in required_columns:
            assert col in columns

        conn.close()

    def test_auction_id_unique_constraint(self, tmp_path):
        """Test that auction_id has UNIQUE constraint"""
        db_path = tmp_path / "test.db"
        conn = sqlite3.connect(str(db_path))
        cursor = conn.cursor()

        cursor.execute("""
            CREATE TABLE IF NOT EXISTS auctions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                auction_id TEXT UNIQUE,
                title TEXT,
                auction_house TEXT,
                current_price REAL,
                estimate_low REAL,
                estimate_high REAL,
                end_date TEXT,
                lot_number TEXT,
                url TEXT,
                search_term TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        # Insert first record
        cursor.execute("""
            INSERT INTO auctions (auction_id, title, auction_house, current_price)
            VALUES (?, ?, ?, ?)
        """, ('test-123', 'Test Bag', 'Christie\'s', 5000))
        conn.commit()

        # Try to insert duplicate
        with pytest.raises(sqlite3.IntegrityError):
            cursor.execute("""
                INSERT INTO auctions (auction_id, title, auction_house, current_price)
                VALUES (?, ?, ?, ?)
            """, ('test-123', 'Another Bag', 'Sotheby\'s', 6000))

        conn.close()


class TestDataInsertion:
    """Test data insertion and retrieval"""

    @pytest.fixture
    def test_db(self, tmp_path):
        """Create test database"""
        db_path = tmp_path / "test.db"
        conn = sqlite3.connect(str(db_path))
        cursor = conn.cursor()

        cursor.execute("""
            CREATE TABLE IF NOT EXISTS auctions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                auction_id TEXT UNIQUE,
                title TEXT,
                auction_house TEXT,
                current_price REAL,
                estimate_low REAL,
                estimate_high REAL,
                end_date TEXT,
                lot_number TEXT,
                url TEXT,
                search_term TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.commit()

        yield conn

        conn.close()

    def test_insert_single_auction(self, test_db):
        """Test inserting a single auction"""
        cursor = test_db.cursor()

        cursor.execute("""
            INSERT INTO auctions
            (auction_id, title, auction_house, current_price, estimate_low,
             estimate_high, end_date, lot_number, url, search_term)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            'test-1', 'Hermes Birkin 30cm', 'Christie\'s', 15000, 18000,
            22000, '2024-01-15', '123', 'https://test.com/123', 'Hermes Birkin'
        ))
        test_db.commit()

        cursor.execute("SELECT COUNT(*) FROM auctions")
        count = cursor.fetchone()[0]

        assert count == 1

    def test_insert_multiple_auctions(self, test_db):
        """Test inserting multiple auctions"""
        cursor = test_db.cursor()

        auctions = [
            ('test-1', 'Bag 1', 'Christie\'s', 5000, 6000, 8000, '2024-01-15', '123', 'https://test.com/1', 'Brand 1'),
            ('test-2', 'Bag 2', 'Sotheby\'s', 10000, 12000, 15000, '2024-01-20', '456', 'https://test.com/2', 'Brand 2'),
            ('test-3', 'Bag 3', 'Phillips', 3000, 4000, 5000, '2024-01-25', '789', 'https://test.com/3', 'Brand 3'),
        ]

        for auction in auctions:
            cursor.execute("""
                INSERT INTO auctions
                (auction_id, title, auction_house, current_price, estimate_low,
                 estimate_high, end_date, lot_number, url, search_term)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, auction)

        test_db.commit()

        cursor.execute("SELECT COUNT(*) FROM auctions")
        count = cursor.fetchone()[0]

        assert count == 3

    def test_insert_or_ignore_duplicate(self, test_db):
        """Test INSERT OR IGNORE for duplicates"""
        cursor = test_db.cursor()

        # Insert first time
        cursor.execute("""
            INSERT OR IGNORE INTO auctions
            (auction_id, title, auction_house, current_price)
            VALUES (?, ?, ?, ?)
        """, ('test-1', 'Bag 1', 'Christie\'s', 5000))
        test_db.commit()

        # Try to insert duplicate - should be ignored
        cursor.execute("""
            INSERT OR IGNORE INTO auctions
            (auction_id, title, auction_house, current_price)
            VALUES (?, ?, ?, ?)
        """, ('test-1', 'Bag 2', 'Sotheby\'s', 10000))
        test_db.commit()

        cursor.execute("SELECT COUNT(*) FROM auctions")
        count = cursor.fetchone()[0]

        # Should still be 1
        assert count == 1

        # Original record should be unchanged
        cursor.execute("SELECT title FROM auctions WHERE auction_id = ?", ('test-1',))
        title = cursor.fetchone()[0]
        assert title == 'Bag 1'

    def test_created_at_timestamp(self, test_db):
        """Test that created_at is set automatically"""
        cursor = test_db.cursor()

        cursor.execute("""
            INSERT INTO auctions
            (auction_id, title, auction_house, current_price)
            VALUES (?, ?, ?, ?)
        """, ('test-1', 'Bag 1', 'Christie\'s', 5000))
        test_db.commit()

        cursor.execute("SELECT created_at FROM auctions WHERE auction_id = ?", ('test-1',))
        created_at = cursor.fetchone()[0]

        assert created_at is not None
        # Should be a valid timestamp
        datetime.strptime(created_at, '%Y-%m-%d %H:%M:%S')


class TestDataRetrieval:
    """Test data retrieval and queries"""

    @pytest.fixture
    def populated_db(self, tmp_path):
        """Create and populate test database"""
        db_path = tmp_path / "test.db"
        conn = sqlite3.connect(str(db_path))
        cursor = conn.cursor()

        cursor.execute("""
            CREATE TABLE IF NOT EXISTS auctions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                auction_id TEXT UNIQUE,
                title TEXT,
                auction_house TEXT,
                current_price REAL,
                estimate_low REAL,
                estimate_high REAL,
                end_date TEXT,
                lot_number TEXT,
                url TEXT,
                search_term TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        # Insert test data
        test_data = [
            ('test-1', 'Hermes Birkin', 'Christie\'s', 15000, 18000, 22000, '2024-01-15', '123', 'https://test.com/1', 'Hermes Birkin'),
            ('test-2', 'Chanel Flap', 'Sotheby\'s', 5500, 6000, 8000, '2024-01-20', '456', 'https://test.com/2', 'Chanel Classic Flap'),
            ('test-3', 'LV Neverfull', 'Heritage', 800, 1200, 1500, '2024-01-25', '789', 'https://test.com/3', 'Louis Vuitton Neverfull'),
            ('test-4', 'Dior Lady Dior', 'Phillips', 3200, 4000, 5000, '2024-02-01', '321', 'https://test.com/4', 'Dior Lady Dior'),
        ]

        for data in test_data:
            cursor.execute("""
                INSERT INTO auctions
                (auction_id, title, auction_house, current_price, estimate_low,
                 estimate_high, end_date, lot_number, url, search_term)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, data)

        conn.commit()
        yield conn
        conn.close()

    def test_retrieve_all_auctions(self, populated_db):
        """Test retrieving all auctions"""
        cursor = populated_db.cursor()
        cursor.execute("SELECT * FROM auctions ORDER BY created_at DESC")
        auctions = cursor.fetchall()

        assert len(auctions) == 4

    def test_filter_by_auction_house(self, populated_db):
        """Test filtering by auction house"""
        cursor = populated_db.cursor()
        cursor.execute("SELECT * FROM auctions WHERE auction_house = ?", ('Christie\'s',))
        auctions = cursor.fetchall()

        assert len(auctions) == 1
        assert auctions[0][3] == 'Christie\'s'  # auction_house column

    def test_filter_by_price_range(self, populated_db):
        """Test filtering by price range"""
        cursor = populated_db.cursor()
        cursor.execute("""
            SELECT * FROM auctions
            WHERE current_price BETWEEN ? AND ?
        """, (1000, 10000))
        auctions = cursor.fetchall()

        assert len(auctions) == 2  # Chanel (5500) and Dior (3200)

    def test_group_by_search_term(self, populated_db):
        """Test grouping by search term"""
        cursor = populated_db.cursor()
        cursor.execute("""
            SELECT search_term, COUNT(*) as count
            FROM auctions
            GROUP BY search_term
            ORDER BY count DESC
        """)
        results = cursor.fetchall()

        assert len(results) == 4
        for result in results:
            assert result[1] > 0  # count should be > 0

    def test_calculate_savings(self, populated_db):
        """Test savings calculation"""
        cursor = populated_db.cursor()
        cursor.execute("""
            SELECT *,
                ((estimate_low - current_price) / estimate_low * 100) as savings_percent,
                (estimate_low - current_price) as savings_amount
            FROM auctions
            WHERE current_price > 0 AND estimate_low > 0
            ORDER BY savings_percent DESC
        """)
        results = cursor.fetchall()

        assert len(results) == 4

        # Verify savings calculations
        for row in results:
            current_price = row[4]
            estimate_low = row[5]
            savings_percent = row[-2]
            savings_amount = row[-1]

            expected_percent = ((estimate_low - current_price) / estimate_low) * 100
            expected_amount = estimate_low - current_price

            assert abs(savings_percent - expected_percent) < 0.01
            assert abs(savings_amount - expected_amount) < 0.01


class TestCSVExport:
    """Test CSV export functionality"""

    def test_csv_file_creation(self, tmp_path):
        """Test that CSV file is created"""
        csv_path = tmp_path / "test.csv"

        import csv
        with open(csv_path, 'w', newline='') as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(['scrape_date', 'auction_id', 'title', 'auction_house'])
            writer.writerow(['2024-01-01', 'test-1', 'Test Bag', 'Christie\'s'])

        assert os.path.exists(csv_path)

    def test_csv_headers(self, tmp_path):
        """Test that CSV has correct headers"""
        csv_path = tmp_path / "test.csv"

        import csv
        headers = [
            'scrape_date', 'auction_id', 'title', 'auction_house',
            'current_price', 'estimate_low', 'estimate_high',
            'end_date', 'lot_number', 'url', 'search_term'
        ]

        with open(csv_path, 'w', newline='') as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(headers)

        with open(csv_path, 'r') as csvfile:
            reader = csv.reader(csvfile)
            csv_headers = next(reader)

        assert csv_headers == headers

    def test_csv_data_export(self, tmp_path):
        """Test exporting data to CSV"""
        csv_path = tmp_path / "test.csv"

        import csv
        test_data = [
            ['2024-01-01', 'test-1', 'Bag 1', 'Christie\'s', '5000', '6000', '8000', '2024-01-15', '123', 'https://test.com/1', 'Brand 1'],
            ['2024-01-01', 'test-2', 'Bag 2', 'Sotheby\'s', '10000', '12000', '15000', '2024-01-20', '456', 'https://test.com/2', 'Brand 2']
        ]

        with open(csv_path, 'w', newline='') as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(['scrape_date', 'auction_id', 'title', 'auction_house', 'current_price', 'estimate_low', 'estimate_high', 'end_date', 'lot_number', 'url', 'search_term'])
            writer.writerows(test_data)

        with open(csv_path, 'r') as csvfile:
            reader = csv.reader(csvfile)
            next(reader)  # Skip header
            rows = list(reader)

        assert len(rows) == 2
        assert rows[0][1] == 'test-1'
        assert rows[1][1] == 'test-2'


class TestLogging:
    """Test logging functionality"""

    def test_log_file_creation(self, tmp_path):
        """Test that log file is created"""
        log_path = tmp_path / "test.log"

        with open(log_path, 'a') as f:
            timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            f.write(f"[{timestamp}] Test log entry\n")

        assert os.path.exists(log_path)

    def test_log_format(self, tmp_path):
        """Test log entry format"""
        log_path = tmp_path / "test.log"

        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        message = "Test log entry"
        log_entry = f"[{timestamp}] {message}\n"

        with open(log_path, 'w') as f:
            f.write(log_entry)

        with open(log_path, 'r') as f:
            content = f.read()

        assert f"[{timestamp}]" in content
        assert message in content


class TestSearchTerms:
    """Test search term configuration"""

    def test_search_terms_list(self):
        """Test that search terms are properly defined"""
        search_terms = [
            "Hermes Birkin",
            "Hermes Kelly",
            "Chanel Classic Flap",
            "Louis Vuitton Neverfull",
            "Dior Lady Dior"
        ]

        assert len(search_terms) > 0
        assert all(isinstance(term, str) for term in search_terms)
        assert all(len(term) > 0 for term in search_terms)


class TestErrorHandling:
    """Test error handling"""

    def test_database_error_handling(self):
        """Test handling of database errors"""
        # Try to connect to non-existent database
        try:
            conn = sqlite3.connect('/nonexistent/path/test.db')
            # Should raise error when trying to write
            cursor = conn.cursor()
            cursor.execute("CREATE TABLE test (id INTEGER)")
            assert False, "Should have raised an error"
        except (sqlite3.Error, OSError):
            assert True

    def test_invalid_data_handling(self, tmp_path):
        """Test handling of invalid data"""
        db_path = tmp_path / "test.db"
        conn = sqlite3.connect(str(db_path))
        cursor = conn.cursor()

        cursor.execute("""
            CREATE TABLE auctions (
                id INTEGER PRIMARY KEY,
                auction_id TEXT UNIQUE,
                current_price REAL
            )
        """)

        # Try to insert invalid price
        try:
            cursor.execute("""
                INSERT INTO auctions (auction_id, current_price)
                VALUES (?, ?)
            """, ('test-1', 'invalid'))
            # SQLite will convert 'invalid' to 0.0, so this won't raise an error
            # We should validate in application code
        except Exception:
            pass

        conn.close()


# Fixtures
@pytest.fixture
def tmp_path(tmpdir):
    """Provide temporary directory"""
    return tmpdir


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