diff options
| -rw-r--r-- | gn3/db/sample_data.py | 32 | ||||
| -rw-r--r-- | tests/unit/db/test_sample_data.py | 21 |
2 files changed, 53 insertions, 0 deletions
diff --git a/gn3/db/sample_data.py b/gn3/db/sample_data.py index ccb03f2..06c3cc5 100644 --- a/gn3/db/sample_data.py +++ b/gn3/db/sample_data.py @@ -11,6 +11,38 @@ _MAP = { } +def __extract_actions(original_data: str, + updated_data: str, + csv_header: str) -> dict: + original_data = original_data.strip().split(",") + updated_data = updated_data.strip().split(",") + csv_header = csv_header.strip().split(",") + result = { + "delete": {"data": [], "csv_header": []}, + "insert": {"data": [], "csv_header": []}, + "update": {"data": [], "csv_header": []}, + } + for _o, _u, _h in zip(original_data, updated_data, csv_header): + if _o == _u: # No change + continue + elif _o and _u == "x": # Deletion + result["delete"]["data"].append(_o) + result["delete"]["csv_header"].append(_h) + elif _o == "x" and _u: # Insert + result["insert"]["data"].append(_u) + result["insert"]["csv_header"].append(_h) + elif _o and _u: # Update + result["update"]["data"].append(_u) + result["update"]["csv_header"].append(_h) + for key, val in result.items(): + if not val["data"]: + result[key] = None + else: + result[key]["data"] = ",".join(result[key]["data"]) + result[key]["csv_header"] = ",".join(result[key]["csv_header"]) + return result + + def get_trait_csv_sample_data(conn: Any, trait_name: int, phenotype_id: int) -> str: """Fetch a trait and return it as a csv string""" diff --git a/tests/unit/db/test_sample_data.py b/tests/unit/db/test_sample_data.py index 4c03dbd..4015ba5 100644 --- a/tests/unit/db/test_sample_data.py +++ b/tests/unit/db/test_sample_data.py @@ -2,6 +2,7 @@ import pytest from gn3.db.sample_data import insert_sample_data from gn3.db.sample_data import delete_sample_data +from gn3.db.sample_data import __extract_actions @pytest.mark.unit_test @@ -73,3 +74,23 @@ def test_delete_sample_data(mocker): (strain_id, "Sex", inbredset_id)), ] cursor.execute.assert_has_calls(calls, any_order=False) + + +@pytest.mark.unit_test +def test_extract_actions(): + assert (__extract_actions(original_data="BXD1,18,x,0,x", + updated_data="BXD1,x,2,1,F", + csv_header="Strain Name,Value,SE,Count,Sex") == + { + "delete": {"data": "18", "csv_header": "Value"}, + "insert": {"data": "2,F", "csv_header": "SE,Sex"}, + "update": {"data": "1", "csv_header": "Count"}, + }) + assert(__extract_actions(original_data="BXD1,18,x,0,x", + updated_data="BXD1,19,2,1,F", + csv_header="Strain Name,Value,SE,Count,Sex") == + { + "delete": None, + "insert": {"data": "2,F", "csv_header": "SE,Sex"}, + "update": {"data": "19,1", "csv_header": "Value,Count"}, + }) |