streamd/test/test_localize.py

from datetime import datetime

from streamer.localize.localize import localize_stream_file
from streamer.localize.localized_shard import LocalizedShard
from streamer.localize.repository_configuration import (
    Dimension,
    Marker,
    MarkerPlacement,
    RepositoryConfiguration,
)
from streamer.parse.shard import Shard, StreamFile

repository_configuration = RepositoryConfiguration(
    dimensions={
        "project": Dimension(
            display_name="Project",
            comment="GTD Project that is being worked on",
            propagate=True,
        ),
        "moment": Dimension(
            display_name="Moment",
            comment="Timestamp this entry was created at",
            propagate=True,
        ),
        "timesheet": Dimension(
            display_name="Timesheet",
            comment="Time Cards for Time Tracking",
            propagate=True,
        ),
    },
    markers={
        "Streamer": Marker(
            display_name="Streamer",
            placements=[
                MarkerPlacement(dimension="project"),
                MarkerPlacement(
                    if_with={"Timesheet"}, dimension="timesheet", value="coding"
                ),
            ],
        ),
        "JobHunting": Marker(
            display_name="JobHunting", placements=[MarkerPlacement(dimension="project")]
        ),
    },
)


class TestLocalize:
    def test_project_simple_stream_file(self):
        stream_file = StreamFile(
            file_name="20250622-121000 Test File.md",
            shard=Shard(start_line=1, end_line=1, markers=["Streamer"]),
        )

        assert localize_stream_file(
            stream_file, repository_configuration
        ) == LocalizedShard(
            moment=datetime(2025, 6, 22, 12, 10, 0, 0),
            markers=["Streamer"],
            tags=[],
            start_line=1,
            end_line=1,
            children=[],
            location={"project": "Streamer", "file": stream_file.file_name},
        )

    def test_timesheet_use_case(self):
        stream_file = StreamFile(
            file_name="20260131-210000 Test File.md",
            shard=Shard(start_line=1, end_line=1, markers=["Timesheet", "Streamer"]),
        )

        assert localize_stream_file(
            stream_file, repository_configuration
        ) == LocalizedShard(
            moment=datetime(2026, 1, 31, 21, 0, 0, 0),
            markers=["Timesheet", "Streamer"],
            tags=[],
            start_line=1,
            end_line=1,
            children=[],
            location={
                "file": stream_file.file_name,
                "project": "Streamer",
                "timesheet": "coding",
            },
        )

    def test_overwrites_true_propagated_dimension_overwrites_existing_value(self):
        config = RepositoryConfiguration(
            dimensions={
                "project": Dimension(display_name="Project", propagate=True),
            },
            markers={
                "A": Marker(
                    display_name="A",
                    placements=[MarkerPlacement(dimension="project", value="a")],
                ),
                "B": Marker(
                    display_name="B",
                    placements=[
                        MarkerPlacement(dimension="project", value="b", overwrites=True)
                    ],
                ),
            },
        )

        stream_file = StreamFile(
            file_name="20260131-210000 Test File.md",
            shard=Shard(start_line=1, end_line=1, markers=["A", "B"]),
        )

        assert localize_stream_file(stream_file, config) == LocalizedShard(
            moment=datetime(2026, 1, 31, 21, 0, 0, 0),
            markers=["A", "B"],
            tags=[],
            start_line=1,
            end_line=1,
            children=[],
            location={"file": stream_file.file_name, "project": "b"},
        )

    def test_overwrites_false_propagated_dimension_does_not_overwrite_existing_value(
        self,
    ):
        config = RepositoryConfiguration(
            dimensions={
                "project": Dimension(display_name="Project", propagate=True),
            },
            markers={
                "A": Marker(
                    display_name="A",
                    placements=[MarkerPlacement(dimension="project", value="a")],
                ),
                "B": Marker(
                    display_name="B",
                    placements=[
                        MarkerPlacement(
                            dimension="project", value="b", overwrites=False
                        )
                    ],
                ),
            },
        )

        stream_file = StreamFile(
            file_name="20260131-210000 Test File.md",
            shard=Shard(start_line=1, end_line=1, markers=["A", "B"]),
        )

        assert localize_stream_file(stream_file, config) == LocalizedShard(
            moment=datetime(2026, 1, 31, 21, 0, 0, 0),
            markers=["A", "B"],
            tags=[],
            start_line=1,
            end_line=1,
            children=[],
            location={"file": stream_file.file_name, "project": "a"},
        )

    def test_overwrites_true_non_propagated_dimension_overwrites_private_value(self):
        config = RepositoryConfiguration(
            dimensions={
                "label": Dimension(display_name="Label", propagate=False),
            },
            markers={
                "A": Marker(
                    display_name="A",
                    placements=[MarkerPlacement(dimension="label", value="a")],
                ),
                "B": Marker(
                    display_name="B",
                    placements=[
                        MarkerPlacement(dimension="label", value="b", overwrites=True)
                    ],
                ),
            },
        )

        stream_file = StreamFile(
            file_name="20260131-210000 Test File.md",
            shard=Shard(start_line=1, end_line=1, markers=["A", "B"]),
        )

        assert localize_stream_file(stream_file, config) == LocalizedShard(
            moment=datetime(2026, 1, 31, 21, 0, 0, 0),
            markers=["A", "B"],
            tags=[],
            start_line=1,
            end_line=1,
            children=[],
            location={"file": stream_file.file_name, "label": "b"},
        )

    def test_overwrites_false_non_propagated_dimension_does_not_overwrite_private_value(
        self,
    ):
        config = RepositoryConfiguration(
            dimensions={
                "label": Dimension(display_name="Label", propagate=False),
            },
            markers={
                "A": Marker(
                    display_name="A",
                    placements=[
                        MarkerPlacement(dimension="label", value="a", overwrites=True)
                    ],
                ),
                "B": Marker(
                    display_name="B",
                    placements=[
                        MarkerPlacement(dimension="label", value="b", overwrites=False)
                    ],
                ),
            },
        )

        stream_file = StreamFile(
            file_name="20260131-210000 Test File.md",
            shard=Shard(start_line=1, end_line=1, markers=["A", "B"]),
        )

        assert localize_stream_file(stream_file, config) == LocalizedShard(
            moment=datetime(2026, 1, 31, 21, 0, 0, 0),
            markers=["A", "B"],
            tags=[],
            start_line=1,
            end_line=1,
            children=[],
            location={"file": stream_file.file_name, "label": "a"},
        )