bearingsinfo/tests/test_processor.py

"""Tests for bearing data processing pipeline."""

import json
import tempfile
from pathlib import Path

import pandas as pd
import pytest

from src.catalog import CatalogManager
from src.config import Config
from src.logger import LoggerSetup, Reporter
from src.parser import DataParser
from src.processor import FileProcessor
from src.registry import Registry
from src.utils import (
    detect_file_type,
    generate_processed_filename,
    make_safe_filename,
    normalize_brand,
    normalize_number,
    normalize_text,
)


@pytest.fixture
def temp_dir():
    """Create temporary directory for tests."""
    with tempfile.TemporaryDirectory() as tmpdir:
        yield Path(tmpdir)


@pytest.fixture
def config_dir(temp_dir):
    """Create test configuration directory."""
    config_dir = temp_dir / "config"
    config_dir.mkdir()

    # Create app.yaml
    app_config = """
paths:
  inbox: inbox
  processed: processed
  error: error
  out: out
  logs: logs

watcher:
  mode: poll
  poll_interval: 1
  process_on_start: true

limits:
  max_file_size_mb: 50
  max_rows: 100000

normalization:
  brand_format: upper
  dimension_replacements:
    "×": "x"
    "х": "x"
  decimal_separator: "."

logging:
  level: INFO
  format: json
  max_size_mb: 10
  backup_count: 5

registry:
  file: out/processed_registry.json
"""
    (config_dir / "app.yaml").write_text(app_config)

    # Create brand_aliases.json
    brand_aliases = {"aliases": {"skf": "SKF", "fag": "FAG", "nsk": "NSK"}}
    (config_dir / "brand_aliases.json").write_text(json.dumps(brand_aliases))

    # Create parsing_rules.json
    parsing_rules = {
        "column_mappings": {
            "Наименование": ["наименование", "Наименование", "name"],
            "Артикул": ["артикул", "Артикул", "designation", "part"],
            "Аналог": ["аналог", "Аналог", "analog"],
            "Бренд": ["бренд", "Бренд", "brand"],
            "d": ["d", "inner diameter"],
            "D": ["D", "outer diameter"],
            "H": ["H", "B", "width"],
            "m": ["m", "weight"],
        },
        "required_fields": {"any_of": ["Артикул", "Наименование"]},
        "dimension_patterns": [
            {
                "regex": r"(?P<d>\d+(?:\.\d+)?)\s*[x×/]\s*(?P<D>\d+(?:\.\d+)?)\s*[x×/]\s*(?P<H>\d+(?:\.\d+)?)",
                "groups": ["d", "D", "H"],
            }
        ],
        "text_normalization": {
            "trim": True,
            "collapse_spaces": True,
            "remove_control_chars": True,
        },
    }
    (config_dir / "parsing_rules.json").write_text(json.dumps(parsing_rules))

    return config_dir


class TestUtils:
    """Test utility functions."""

    def test_normalize_text(self):
        """Test text normalization."""
        assert normalize_text("  test  ") == "test"
        assert normalize_text("test   multiple   spaces") == "test multiple spaces"
        assert normalize_text("test×100") == "test×100"

    def test_normalize_number(self):
        """Test number normalization."""
        assert normalize_number(10) == 10.0
        assert normalize_number("10") == 10.0
        assert normalize_number("10.5") == 10.5
        assert normalize_number("10,5") == 10.5
        assert normalize_number("") is None
        assert normalize_number(None) is None

    def test_normalize_brand(self):
        """Test brand normalization."""
        aliases = {"skf": "SKF", "fag": "FAG"}
        assert normalize_brand("skf", aliases, "upper") == "SKF"
        assert normalize_brand("SKF", aliases, "upper") == "SKF"
        assert normalize_brand("nsk", {}, "upper") == "NSK"

    def test_make_safe_filename(self):
        """Test safe filename generation."""
        assert make_safe_filename("test file.csv") == "test_file"
        assert make_safe_filename("Test-File_123") == "Test-File_123"
        assert make_safe_filename("файл@#$.csv") == "файл"

    def test_generate_processed_filename(self):
        """Test processed filename generation."""
        name = generate_processed_filename("test.csv", 100, "a" * 64, is_error=False)
        assert "__test__100__" in name
        assert name.endswith(".csv")

        error_name = generate_processed_filename("test.csv", 0, "a" * 64, is_error=True, error_code="PARSE")
        assert "ERROR_PARSE" in error_name

    def test_detect_file_type(self):
        """Test file type detection."""
        assert detect_file_type(Path("test.csv")) == "csv"
        assert detect_file_type(Path("test.xlsx")) == "xlsx"
        assert detect_file_type(Path("test.json")) == "json"
        assert detect_file_type(Path("test.txt")) == "txt"
        assert detect_file_type(Path("test.xyz")) == "unknown"


class TestRegistry:
    """Test registry management."""

    def test_registry_operations(self, temp_dir):
        """Test registry add and check operations."""
        registry_file = temp_dir / "registry.json"
        registry = Registry(registry_file)

        # Should not be processed initially
        assert not registry.is_processed("hash123")

        # Add entry
        registry.add_entry(
            file_hash="hash123",
            original_name="test.csv",
            processed_name="20240101_120000__test__100__hash123.csv",
            n_records=100,
        )

        # Should now be processed
        assert registry.is_processed("hash123")

        # Get entry
        entry = registry.get_entry("hash123")
        assert entry["original_name"] == "test.csv"
        assert entry["n_records"] == 100


class TestParser:
    """Test data parsing."""

    def test_parse_csv(self, temp_dir, config_dir):
        """Test CSV parsing."""
        # Create test CSV
        csv_file = temp_dir / "test.csv"
        df = pd.DataFrame(
            {"артикул": ["6200", "6201"], "бренд": ["SKF", "FAG"], "d": [10, 12], "D": [30, 32], "H": [9, 10]}
        )
        df.to_csv(csv_file, index=False)

        # Parse
        config = Config(config_dir)
        parser = DataParser(config.load_parsing_rules(), {})

        result = parser.parse_file(csv_file, "csv")
        assert len(result) == 2
        assert "артикул" in result.columns

    def test_normalize_columns(self, config_dir):
        """Test column normalization."""
        config = Config(config_dir)
        parser = DataParser(config.load_parsing_rules(), {})

        df = pd.DataFrame({"part": ["6200"], "brand": ["SKF"], "inner diameter": [10]})

        normalized = parser.normalize_columns(df)
        # Should be mapped to target columns (title case)
        assert "Артикул" in normalized.columns
        assert "Бренд" in normalized.columns
        assert "d" in normalized.columns


class TestCatalog:
    """Test catalog management."""

    def test_add_records_deduplication(self, temp_dir, config_dir):
        """Test adding records with deduplication."""
        catalog_csv = temp_dir / "catalog.csv"
        catalog_json = temp_dir / "catalog.json"

        config = Config(config_dir)
        catalog = CatalogManager(
            catalog_csv=catalog_csv,
            catalog_json=catalog_json,
            brand_aliases=config.load_brand_aliases(),
            normalization_config={"brand_format": "upper"},
        )

        # Add first record
        df1 = pd.DataFrame({"Артикул": ["6200"], "Бренд": ["SKF"], "d": [10], "D": [30], "H": [9]})

        n_added, n_skipped, n_conflicts, _ = catalog.add_records(df1)
        assert n_added == 1
        assert n_skipped == 0

        # Add duplicate - should skip
        n_added, n_skipped, n_conflicts, _ = catalog.add_records(df1)
        assert n_added == 0
        assert n_skipped == 1

    def test_add_records_conflict(self, temp_dir, config_dir):
        """Test adding records with dimension conflicts."""
        catalog_csv = temp_dir / "catalog.csv"
        catalog_json = temp_dir / "catalog.json"

        config = Config(config_dir)
        catalog = CatalogManager(
            catalog_csv=catalog_csv,
            catalog_json=catalog_json,
            brand_aliases=config.load_brand_aliases(),
            normalization_config={"brand_format": "upper"},
        )

        # Add first record
        df1 = pd.DataFrame({"Артикул": ["6200"], "Бренд": ["SKF"], "d": [10], "D": [30], "H": [9]})
        catalog.add_records(df1)

        # Add record with same article/brand but different dimensions
        df2 = pd.DataFrame({"Артикул": ["6200"], "Бренд": ["SKF"], "d": [10], "D": [30], "H": [10]})  # Different!

        n_added, n_skipped, n_conflicts, conflicts = catalog.add_records(df2)
        assert n_added == 1  # Should add as separate entry
        assert n_conflicts == 1  # But mark as conflict


class TestProcessor:
    """Test file processor."""

    def test_process_csv_file(self, temp_dir, config_dir):
        """Test processing a CSV file."""
        # Setup directories
        inbox = temp_dir / "inbox"
        processed = temp_dir / "processed"
        error = temp_dir / "error"
        out = temp_dir / "out"
        logs = temp_dir / "logs"

        for d in [inbox, processed, error, out, logs]:
            d.mkdir()

        # Create test CSV in inbox
        csv_file = inbox / "test.csv"
        df = pd.DataFrame(
            {
                "артикул": ["6200", "6201", "6202"],
                "бренд": ["SKF", "FAG", "NSK"],
                "d": [10, 12, 15],
                "D": [30, 32, 35],
                "H": [9, 10, 11],
            }
        )
        df.to_csv(csv_file, index=False)

        # Setup config to use temp directories
        config = Config(config_dir)
        log_file = logs / "app.log"
        logger = LoggerSetup.setup(log_file, "json", "INFO")

        # Change working directory context
        import os

        old_cwd = os.getcwd()
        try:
            os.chdir(temp_dir)

            processor = FileProcessor(config, logger)

            # Process file
            status, n_records = processor.process_file(csv_file)

            # Check results
            assert status == "success"
            assert n_records == 3

            # File should be moved to processed
            assert not csv_file.exists()
            assert len(list(processed.glob("*.csv"))) == 1

            # Catalog should exist
            assert (out / "catalog_target.csv").exists()
            assert (out / "catalog_target.json").exists()

            # Report should exist
            assert (out / "run_report.ndjson").exists()

        finally:
            os.chdir(old_cwd)

    def test_idempotent_processing(self, temp_dir, config_dir):
        """Test that processing same file twice doesn't duplicate records."""
        # Setup directories
        inbox = temp_dir / "inbox"
        processed = temp_dir / "processed"
        out = temp_dir / "out"
        logs = temp_dir / "logs"

        for d in [inbox, processed, out, logs]:
            d.mkdir()

        # Create test CSV properly
        csv_file1 = inbox / "test1.csv"
        df = pd.DataFrame(
            {"артикул": ["6200", "6201"], "бренд": ["SKF", "FAG"], "d": [10, 12], "D": [30, 32], "H": [9, 10]}
        )
        df.to_csv(csv_file1, index=False)

        # Setup and process first file
        config = Config(config_dir)
        log_file = logs / "app.log"
        logger = LoggerSetup.setup(log_file, "json", "INFO")

        import os

        old_cwd = os.getcwd()
        try:
            os.chdir(temp_dir)

            processor = FileProcessor(config, logger)
            processor.process_file(csv_file1)

            # Check catalog
            catalog_df = pd.read_csv(out / "catalog_target.csv")
            assert len(catalog_df) == 2

            # Create identical file with different name
            csv_file2 = inbox / "test2.csv"
            df.to_csv(csv_file2, index=False)

            # Process second file (same content)
            status, _ = processor.process_file(csv_file2)

            # Should be skipped as duplicate
            assert status == "skipped_duplicate"

            # Catalog should still have only 2 records
            catalog_df = pd.read_csv(out / "catalog_target.csv")
            assert len(catalog_df) == 2

        finally:
            os.chdir(old_cwd)


class TestReporter:
    """Test reporter."""

    def test_write_report(self, temp_dir):
        """Test writing NDJSON report."""
        report_file = temp_dir / "report.ndjson"
        reporter = Reporter(report_file)

        reporter.write_report(
            filename="test.csv", file_hash="abc123", status="success", n_rows=10, n_added=8, n_skipped=2, n_conflicts=0
        )

        # Check report exists and is valid NDJSON
        assert report_file.exists()

        with open(report_file) as f:
            line = f.readline()
            data = json.loads(line)

            assert data["filename"] == "test.csv"
            assert data["status"] == "success"
            assert data["n_added"] == 8