DHIS-21957 Skip data values for closed orgunits before dataValueSets POST

app/cli.py

@@ -13,6 +13,8 @@
 from app.core.config_loader import ConfigManager
 from app.core.api_utils import Dhis2ApiUtils
 
+from app.core.org_unit_filter import filter_and_fetch_closed_org_units
+
 
 def _format_duration(delta) -> str:
     total = delta.total_seconds()
@@ -72,7 +74,7 @@
             return await analyzer.run_stage(stage, session, semaphore)
 
         except Exception as e:
             logging.error(f"Error running stage '{stage.get('name', '<unnamed>')}': {e}")
             return []
 
     async def run_all_stages(self):
@@ -121,6 +123,17 @@
                 logging.warning(msg)
                 errors.append(msg)
 
+        if upserts:
+            upserts, dropped_closed = await filter_and_fetch_closed_org_units(
+                self.api_utils, upserts, session
+            )
+            if dropped_closed:
+                logging.warning(f"Skipping {len(dropped_closed)} data value(s) for closed org units")
+                errors.extend(
+                    f"Skipped (org unit closed): {dv['orgUnit']}/{dv['period']}"
+                    for dv in dropped_closed
+                )
+
         import_summary = None
         delete_import_summary = None
         integrity_import_summary = None
@@ -142,7 +155,7 @@
                     group_summaries.append(Dhis2ApiUtils.parse_import_summary(response))
                 import_summary = self._merge_import_summaries(*group_summaries) if group_summaries else None
             except Exception as post_err:
                 logging.error(f"Error posting data values: {post_err}")
                 errors.append(f"Post failed: {post_err}")
 
         if deletes:
@@ -153,7 +166,7 @@
                 )
                 delete_import_summary = Dhis2ApiUtils.parse_import_summary(response)
             except Exception as delete_err:
                 logging.error(f"Error deleting data values: {delete_err}")
                 errors.append(f"Delete failed: {delete_err}")
 
         for payload in integrity_payloads:
@@ -165,7 +178,7 @@
                 response = await self.api_utils.post_data_value_set(payload, session)
                 integrity_import_summary = Dhis2ApiUtils.parse_import_summary(response)
             except Exception as integrity_err:
                 logging.error(f"Error posting integrity data values: {integrity_err}")
                 errors.append(f"Integrity post failed: {integrity_err}")
 
         combined_import_summary = self._merge_import_summaries(

app/core/api_utils.py

@@ -71,6 +71,34 @@ async def get_organisation_units_at_level(self, level, session, semaphore):
                 data = await response.json()
                 return [ou['id'] for ou in data['organisationUnits']]
 
+    async def get_organisation_unit_dates_bulk(self, org_unit_ids, session, chunk_size=200):
+        """
+        Fetch the openingDate and closedDate for a batch of org units.
+
+        Splits into multiple requests if org_unit_ids is large, since DHIS2's
+        filter query string has a practical length limit.
+
+        Returns a dict keyed by org_unit_id as follows:
+        {
+            'DiszpKrYNg8': {'openingDate': '2003-01-01', 'closedDate': None},
+            'BoskqZsekw8': {'openingDate': '2015-06-16', 'closedDate': '2024-03-01'},
+        }
+        """
+        org_unit_ids = list(org_unit_ids)
+        all_dates = {}
+
+        for i in range(0, len(org_unit_ids), chunk_size):
+            chunk = org_unit_ids[i:i + chunk_size]
+            ids_filter = ','.join(chunk)
+            url = (f'{self.base_url}/api/organisationUnits.json'
+                   f'?filter=id:in:[{ids_filter}]&fields=id,openingDate,closedDate&paging=false')
+            async with session.get(url) as response:
+                response.raise_for_status()
+                data = await response.json()
+                all_dates.update({ou['id']: ou for ou in data['organisationUnits']})
+
+        return all_dates
+
     async def fetch_datavalue_sets(self, query_params, session):
         url = f'{self.base_url}/api/dataValueSets.json'
         async with session.get(url, params=query_params) as response:

app/core/org_unit_filter.py

@@ -0,0 +1,50 @@
+# app/core/org_unit_filter.py
+
+from datetime import datetime
+from app.core.period_utils import Dhis2PeriodUtils
+
+def _parse_date(date_str):
+    if not date_str:
+        return None
+    return datetime.strptime(date_str[:10], "%Y-%m-%d")
+
+def should_keep_data_value(dv, opening_date, closed_date, period_utils=None):
+    """
+    Keep a data value if its period overlaps at all with the org unit's open window .
+    A period that only partially overlaps (e.g. the org unit opens or closes mid-period)
+    still counts as kept, because there could be valid data produced in that period
+    """
+
+    period_utils = period_utils or Dhis2PeriodUtils()
+
+    opening = _parse_date(opening_date)
+    closed = _parse_date(closed_date)
+
+    period_start = period_utils.get_start_date_from_period(dv['period'])
+    period_end = period_utils.get_end_date_from_period(dv['period'])
+
+    if opening and period_end < opening:
+        return False
+    if closed and period_start > closed:
+        return False
+    return True
+
+async def filter_and_fetch_closed_org_units(api_utils, data_values, session):
+    """Fetch org unit dates for all org units referenced in data_values,
+    then keep only the values whose period overlaps that org unit's open
+    window (per should_keep_data_value)."""
+    if not data_values:
+        return [], []
+
+    org_unit_ids = {dv['orgUnit'] for dv in data_values}
+    ou_dates = await api_utils.get_organisation_unit_dates_bulk(org_unit_ids, session)
+
+    kept, dropped = [], []
+    for dv in data_values:
+        ou = ou_dates.get(dv['orgUnit'], {})
+        if should_keep_data_value(dv, ou.get('openingDate'), ou.get('closedDate')):
+            kept.append(dv)
+        else:
+            dropped.append(dv)
+
+    return kept, dropped

pyproject.toml

@@ -1,3 +1,3 @@
 # pyproject.toml
 [build-system]
 requires = ["setuptools>=68", "wheel"]
@@ -26,6 +26,7 @@
 # Install with: pip install -e .[dev]
 dev = [
   "pytest>=8.3,<9",
+  "pytest-asyncio>=0.24,<1.0",
   "bump-my-version",
 ]
 

pytest.ini

@@ -3,4 +3,5 @@
 pythonpath = app
 testpaths = tests
 addopts = -v
+asyncio_mode = auto
 

tests/test_api_utils.py

@@ -1,8 +1,83 @@
-from unittest.mock import patch, MagicMock
+import pytest
+from unittest.mock import AsyncMock, patch, MagicMock
 import requests
 
 from app.core.api_utils import Dhis2ApiUtils
 
+def _mock_session_with_responses(payloads):
+    """An aiohttp.ClientSession stand-in whose .get() returns a different
+    payload on each successive call, in the order given. No real network
+    access occurs."""
+    session = MagicMock()
+    call_urls = []
+
+    def fake_get(url, *args, **kwargs):
+        call_urls.append(url)
+        payload = payloads[len(call_urls) - 1]
+
+        response = AsyncMock()
+        response.raise_for_status = MagicMock()
+        response.json = AsyncMock(return_value=payload)
+
+        cm = AsyncMock()
+        cm.__aenter__.return_value = response
+        return cm
+
+    session.get = MagicMock(side_effect=fake_get)
+    session.call_urls = call_urls
+    return session
+
+
+async def test_get_organisation_unit_dates_bulk_chunks_large_id_lists():
+    """With more org unit IDs than chunk_size, the method should split into
+    multiple requests and merge all results into one dict."""
+    api = _make_utils()
+
+    org_unit_ids = [f'ou{i:09d}' for i in range(5)]  # 5 fake IDs
+    chunk_size = 2  # forces 3 chunks: [2, 2, 1]
+
+    responses = [
+        {'organisationUnits': [
+            {'id': 'ou000000000', 'openingDate': '2020-01-01', 'closedDate': None},
+            {'id': 'ou000000001', 'openingDate': '2020-01-01', 'closedDate': None},
+        ]},
+        {'organisationUnits': [
+            {'id': 'ou000000002', 'openingDate': '2020-01-01', 'closedDate': None},
+            {'id': 'ou000000003', 'openingDate': '2020-01-01', 'closedDate': None},
+        ]},
+        {'organisationUnits': [
+            {'id': 'ou000000004', 'openingDate': '2020-01-01', 'closedDate': None},
+        ]},
+    ]
+    mock_session = _mock_session_with_responses(responses)
+
+    result = await api.get_organisation_unit_dates_bulk(org_unit_ids, mock_session, chunk_size=chunk_size)
+
+    # Three separate requests were made, not one giant one
+    assert mock_session.get.call_count == 3
+
+    # All five org units ended up in the merged result
+    assert set(result.keys()) == set(org_unit_ids)
+    assert result['ou000000004']['openingDate'] == '2020-01-01'
+
+
+async def test_get_organisation_unit_dates_bulk_single_chunk_when_small():
+    """With fewer IDs than chunk_size, only one request should be made."""
+    api = _make_utils()
+
+    org_unit_ids = ['hZpaU5uFSDm']
+    mock_session = _mock_session_with_responses([
+        {'organisationUnits': [
+            {'id': 'hZpaU5uFSDm', 'openingDate': '2003-01-16', 'closedDate': '2026-02-16'},
+        ]}
+    ])
+
+    result = await api.get_organisation_unit_dates_bulk(org_unit_ids, mock_session, chunk_size=200)
+
+    assert mock_session.get.call_count == 1
+    assert result['hZpaU5uFSDm']['closedDate'] == '2026-02-16'
+
+
 
 def _make_utils():
     return Dhis2ApiUtils('https://dhis2.example.org', 'fake-token')

tests/test_org_unit_filter.py

@@ -0,0 +1,109 @@
+# tests/test_org_unit_filter.py
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+
+from app.core.api_utils import Dhis2ApiUtils
+from app.core.org_unit_filter import should_keep_data_value, filter_and_fetch_closed_org_units
+
+@pytest.fixture
+def mock_session():
+    """
+    An aiohttp.ClientSession stand-in whose .get() is an async context manager. Call the
+    .set_response(json_payload) to control what it returns. No real network access is made
+    anywhere in this module
+    """
+    response = AsyncMock()
+    response.raise_for_status = MagicMock()
+
+    get_cm = AsyncMock()
+    get_cm.__aenter__.return_value = response
+
+    session = MagicMock()
+    session.get = MagicMock(return_value=get_cm)
+    session.set_response = lambda payload:setattr(response, 'json', AsyncMock(return_value=payload))
+
+    return session
+
+# Representative periods spanning the window, including 202601 specifically
+# to catch the "opening date falls inside this period" boundary.
+PERIODS = ['202507', '202512', '202601', '202602', '202603', '202606']
+
+
+def kept_periods(opening_date, closed_date):
+    """Helper: which of PERIODS does should_keep_data_value keep, given one
+    opening/closed date pair? Returns a set of period strings."""
+    return {
+        p for p in PERIODS
+        if should_keep_data_value({'orgUnit': 'hZpaU5uFSDm', 'period': p}, opening_date, closed_date)
+    }
+
+
+def test_closed_before_window_drops_everything():
+    """Org unit was already closed before the window even starts -> nothing
+    in the window should be kept."""
+    opening_date = '2003-01-16'
+    closed_date = '2025-02-16'  # before window_start (202507)
+
+    assert kept_periods(opening_date, closed_date) == set()
+
+
+def test_closes_during_window_keeps_only_periods_before_closure():
+    """Org unit closes partway through the window -> periods up to closure
+    kept, periods after dropped."""
+    opening_date = '2003-01-16'
+    closed_date = '2026-02-16'  # closes during the window, mid-February
+
+    assert kept_periods(opening_date, closed_date) == {'202507', '202512', '202601', '202602'}
+
+
+def test_opens_and_closes_during_window_keeps_only_the_overlap():
+    """Org unit both opens and closes inside the window -> only periods
+    overlapping [opening_date, closed_date] are kept, including the exact
+    opening month."""
+    opening_date = '2025-07-16'  # during window, mid-July -> falls in period 202507
+    closed_date = '2026-02-16'   # during window, mid-February
+
+    assert kept_periods(opening_date, closed_date) == {'202507', '202512', '202601', '202602'}
+
+
+def test_opens_during_window_closes_after_window_keeps_periods_from_opening_onward():
+    """Org unit opens partway through the window, and closedDate is set in
+    the future beyond the window end (UI allows future dates) -> periods
+    from the opening month through the end of the window are kept."""
+    opening_date = '2026-01-16'  # during window, mid-January -> falls in period 202601
+    closed_date = '2028-02-16'   # after window_end (future-dated)
+
+    assert kept_periods(opening_date, closed_date) == {'202601', '202602', '202603', '202606'}
+
+
+def test_opens_after_window_drops_everything():
+    """Org unit doesn't open until after the window ends entirely (also
+    future-dated) -> nothing in the window should be kept."""
+    opening_date = '2028-01-16'
+    closed_date = '2030-02-16'
+
+    assert kept_periods(opening_date, closed_date) == set()
+
+
+
+
+async def test_closed_org_unit_excluded_via_bulk_lookup(mock_session):
+    """
+    The bulk org unit lookup used with should_keep_data_value together should
+    exclude a post-closure data value before it gets posted
+    """
+    mock_session.set_response({'organisationUnits': [
+        {'id': 'hZpaU5uFSDm', 'openingDate': '2003-01-16', 'closedDate': '2026-02-16'},
+    ]})
+
+    before_closure = {'orgUnit': 'hZpaU5uFSDm', 'period': '202512', 'value': '1'}
+    after_closure = {'orgUnit': 'hZpaU5uFSDm', 'period': '202604', 'value': '2'}
+
+    kept, dropped = await filter_and_fetch_closed_org_units(
+        api_utils=Dhis2ApiUtils('https://dummy-url.com', 'dummy-token'),
+        data_values=[before_closure, after_closure],
+        session=mock_session,
+    )
+
+    assert kept == [before_closure]
+    assert dropped == [after_closure]