from datetime import datetime from streamd.localize.localize import localize_stream_file from streamd.localize.localized_shard import LocalizedShard from streamd.localize.repository_configuration import ( Dimension, Marker, MarkerPlacement, RepositoryConfiguration, ) from streamd.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={ "Streamd": Marker( display_name="Streamd", 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=["Streamd"]), ) assert localize_stream_file( stream_file, repository_configuration ) == LocalizedShard( moment=datetime(2025, 6, 22, 12, 10, 0, 0), markers=["Streamd"], tags=[], start_line=1, end_line=1, children=[], location={"project": "Streamd", "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", "Streamd"]), ) assert localize_stream_file( stream_file, repository_configuration ) == LocalizedShard( moment=datetime(2026, 1, 31, 21, 0, 0, 0), markers=["Timesheet", "Streamd"], tags=[], start_line=1, end_line=1, children=[], location={ "file": stream_file.file_name, "project": "Streamd", "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"}, )