tests/test_prepare_features.py - master - maruga/mlops

tests/test_prepare_features.py
215 строк8 KB
s546371
modified: .gitlab-ci.yml
22 мар 2026, 09:51
22 мар 2026, 09:5127798b8
from pathlib import Path

import pandas as pd
import pytest

from prepare_features import (
    build_target,
    compute_item_popularity,
    load_events,
    load_item_properties_from_files,
    prepare,
)
from src.config import DataMap


class TestPrepareFeatures:
    """Класс с юнит- и интеграционными тестами предобработки данных"""

    @pytest.mark.parametrize(
        "sample_events_csv",
        [
            """1609459260000,purchase,1,100,123
        1609459320000,transaction,2,200,124
        1609459380000,addtocart,3,300,
        1609459440000,view,4,400,"""
        ],
        indirect=True,
    )
    def test_load_events_basic(self, sample_events_csv):
        """Основные преобразования"""
        df = load_events(sample_events_csv)
        assert isinstance(df, pd.DataFrame)
        assert len(df) == 4
        assert "timestamp" in df.columns
        assert pd.api.types.is_datetime64_any_dtype(df["timestamp"])

    @pytest.mark.parametrize(
        "sample_events_csv", ["1609459200000,view,1,100,"], indirect=True
    )
    def test_does_not_purchase_detection(self, sample_events_csv):
        """Проверяет определение покупок"""
        df = load_events(sample_events_csv)
        assert not df.loc[0, "is_purchase_event"]

    def _base_test_purchase_detection(self, sample_events_csv):
        """Базовая проверка определение покупки"""
        df = load_events(sample_events_csv)
        assert df.loc[0, "is_purchase_event"]

    @pytest.mark.parametrize(
        "sample_events_csv", ["1609459260000,purchase,1,100,123"], indirect=True
    )
    def test_purchase_by_event_and_id_detection(self, sample_events_csv):
        """Проверяет определение покупок"""
        self._base_test_purchase_detection(sample_events_csv)

    @pytest.mark.parametrize(
        "sample_events_csv", ["1609459320000,transaction,2,200,124"], indirect=True
    )
    def test_purchase_by_id_detection(self, sample_events_csv):
        """Проверяет определение покупок"""
        self._base_test_purchase_detection(sample_events_csv)

    @pytest.mark.parametrize(
        "sample_events_csv", ["1609459380000,purchase,3,300,"], indirect=True
    )
    def test_purchase_by_event_detection(self, sample_events_csv):
        """Проверяет определение покупок"""
        self._base_test_purchase_detection(sample_events_csv)

    def test_load_item_properties_latest_values(self, tmp_path):
        """Проверяет выбор последних значений"""
        csv_content = """itemid,property,value
        100,categoryid,electronics
        100,categoryid,updated_category
        100,available,5
        100,available,10
        100,available,15"""
        file_path = tmp_path / "props.csv"
        file_path.write_text(csv_content)

        df = load_item_properties_from_files([Path(file_path)])
        assert df.loc[0, "categoryid"] == "updated_category"
        assert df.loc[0, "available"] == "15"

    def test_compute_popularity_basic(self) -> None:
        """Проверяет расчет ctr"""
        data = {
            "event": ["view", "purchase", "view", "purchase", "view"],
            "itemid": [1, 1, 2, 2, 3],
            "is_purchase_event": [False, True, False, False, False],
        }
        events = pd.DataFrame(data)

        result = compute_item_popularity(events)
        assert DataMap.views in result.columns
        assert DataMap.purchases in result.columns
        assert DataMap.ctr in result.columns

        item1 = result[result["itemid"] == 1].iloc[0]
        assert item1[DataMap.views] == 1
        assert item1[DataMap.purchases] == 1
        assert item1[DataMap.ctr] == 1.0

        item2 = result[result["itemid"] == 2].iloc[0]
        assert item2[DataMap.views] == 1
        assert item2[DataMap.purchases] == 0
        assert item2[DataMap.ctr] == 0.0

        item3 = result[result["itemid"] == 3].iloc[0]
        assert item3[DataMap.views] == 1
        assert item3[DataMap.purchases] == 0
        assert item3[DataMap.ctr] == 0.0

    def test_build_target_basic(self) -> None:
        """Проверяет соответствие цели, если покупка попала в границы временного промежутка"""
        views = pd.DataFrame(
            {
                "timestamp": pd.to_datetime(["2025-01-01 10:00", "2025-01-01 10:00"]),
                "visitorid": [1, 2],
                "itemid": [100, 200],
            }
        )

        purchases = pd.DataFrame(
            {
                "timestamp": pd.to_datetime(["2025-01-01 10:01", "2025-01-02 10:00"]),
                "visitorid": [1, 2],
                "itemid": [100, 200],
            }
        )

        result = build_target(views, purchases, window_hours=24)

        assert result.loc[0, DataMap.target] == 1
        assert result.loc[1, DataMap.target] == 1

    def test_build_target_not_into_window(self) -> None:
        """Проверяет соответствие цели, если покупка НЕ попала на границу временного промежутка"""
        views = pd.DataFrame(
            {
                "timestamp": pd.to_datetime(["2025-01-01 10:00"]),
                "visitorid": [1],
                "itemid": [100],
            }
        )

        purchases = pd.DataFrame(
            {
                "timestamp": pd.to_datetime(["2025-01-01 22:01"]),
                "visitorid": [1],
                "itemid": [100],
            }
        )

        result = build_target(views, purchases, window_hours=12)
        assert result.loc[0, DataMap.target] == 0

    def test_build_target_no_purchase(self) -> None:
        """Проверяет цель при отсутствии покупки"""
        views = pd.DataFrame(
            {
                "timestamp": pd.to_datetime(["2025-01-01 10:00"]),
                "visitorid": [1],
                "itemid": [100],
            }
        )

        purchases = pd.DataFrame(
            {"timestamp": pd.to_datetime([]), "visitorid": [], "itemid": []}
        )

        result = build_target(views, purchases)
        assert result.loc[0, DataMap.target] == 0

    def test_prepare_end_to_end(self, tmp_path):
        """Интеграционный тест"""
        events_content = """timestamp,event,visitorid,itemid,transactionid
        1609459200000,view,1,100,
        1609459260000,purchase,1,100,123
        1609459320000,view,2,200,
        1609459380000,purchase,2,200,124
        """

        props_content = """itemid,property,value
        100,categoryid,electronics
        100,available,10
        200,categoryid,books
        200,available,5
        """

        events_path = tmp_path / "events.csv"
        props_path = tmp_path / "props.csv"
        out_item_path = tmp_path / "item_feats.parquet"
        out_train_path = tmp_path / "train.parquet"

        events_path.write_text(events_content)
        props_path.write_text(props_content)

        prepare(
            Path(events_path),
            [Path(props_path)],
            str(out_item_path),
            str(out_train_path),
            window_hours=24,
        )

        assert out_item_path.exists()
        assert out_train_path.exists()
        assert not pd.read_parquet(out_item_path).empty

        train_parquet = pd.read_parquet(out_train_path)
        assert DataMap.target in train_parquet.columns
        for col in DataMap.features:
            assert col in train_parquet.columns