# (c) cavaliba.com - tests / cli "cavaliba load" CSV path import os import re import tempfile from io import StringIO import yaml from django.core.management import call_command from django.test import TestCase import app_home.cache as cache from app_data.data import Instance from app_data.loader import load_broker from tests.helper import add_schema class CliLoadCsvTest(TestCase): """`cavaliba load file.csv [--pipeline X]` goes through the same load_file_csv() as the web Import Tool - these lock in that the CLI path still works after that function was extended (optional pipeline, per-row "schema" column, "^" multi-value splitting).""" def setUp(self): cache.clear() add_schema( classname="clicsvtest", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, "tags": { "dataformat": "string", "displayname": "Tags", "cardinal_max": 0, }, }, ) datalist = yaml.safe_load(""" - classname: _pipeline keyname: pipeline_clicsvtest displayname: pipeline_clicsvtest is_enabled: True content: | classname: clicsvtest keyfield: keyname tasks: [] """) load_broker(datalist=datalist, aaa={"perms": ["p_pipeline_create"]}) def _write_csv(self, content): fd, path = tempfile.mkstemp(suffix=".csv") with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(content) return path def test_csv_load_via_pipeline_creates_instance(self): """`--pipeline` still supplies classname/keyfield exactly as before this refactor - no --schema equivalent flag needed.""" path = self._write_csv("keyname,mystring\nrow1,hello\n") try: output = StringIO() call_command( "cavaliba", "load", path, "--pipeline", "pipeline_clicsvtest", verbosity=2, stdout=output, ) finally: os.remove(path) self.assertIn("load done", output.getvalue()) instance = Instance.from_keyname(classname="clicsvtest", keyname="row1", expand=True) self.assertIsNotNone(instance) self.assertEqual(instance.fields["mystring"].value[0], "hello") def test_csv_load_without_pipeline_uses_row_schema_column(self): """New capability: a per-row "schema" column lets `cavaliba load` import a CSV with no --pipeline flag at all.""" path = self._write_csv("keyname,schema,mystring\nrow2,clicsvtest,hello\n") try: output = StringIO() call_command("cavaliba", "load", path, verbosity=2, stdout=output) finally: os.remove(path) self.assertIn("load done", output.getvalue()) instance = Instance.from_keyname(classname="clicsvtest", keyname="row2", expand=True) self.assertIsNotNone(instance) def test_csv_load_without_pipeline_or_schema_column_imports_nothing(self): """Without --pipeline and without a "schema" column, every row is skipped (missing classname) - load_file_csv now returns an empty list instead of hard-aborting, and the CLI reports it as "no data" rather than crashing.""" path = self._write_csv("keyname,mystring\nrow3,hello\n") try: output = StringIO() err = StringIO() call_command("cavaliba", "load", path, verbosity=2, stdout=output, stderr=err) finally: os.remove(path) self.assertIn("load done", output.getvalue()) self.assertIn("SKIP - no data in", err.getvalue()) instance = Instance.from_keyname(classname="clicsvtest", keyname="row3", expand=False) self.assertIsNone(instance) def test_csv_load_progress_flag_prints_progress(self): """--progress must actually print progress lines during the load loop, not just echo the flag's value once.""" content = "keyname,mystring\n" + "\n".join(f"row{i},hello{i}" for i in range(1, 4)) + "\n" path = self._write_csv(content) try: output = StringIO() call_command( "cavaliba", "load", path, "--pipeline", "pipeline_clicsvtest", "--progress", verbosity=2, stdout=output, ) finally: os.remove(path) rendered = output.getvalue() self.assertIn("load done", rendered) self.assertRegex(rendered, r"progress:\s*3/3\s*") def test_csv_load_without_progress_flag_prints_no_progress(self): """Without --progress, no progress lines should appear (default behavior).""" path = self._write_csv("keyname,mystring\nrow1,hello\n") try: output = StringIO() call_command( "cavaliba", "load", path, "--pipeline", "pipeline_clicsvtest", verbosity=2, stdout=output, ) finally: os.remove(path) self.assertNotIn("progress: 1/1", output.getvalue()) def test_csv_load_multivalue_field_splits_on_caret(self): """ "^"-joined multi-value cells split into a list through the CLI path too - this logic lives in load_file_csv() itself, shared with the web Import Tool.""" path = self._write_csv("keyname,schema,tags\nrow4,clicsvtest,a^b^c\n") try: output = StringIO() call_command("cavaliba", "load", path, verbosity=2, stdout=output) finally: os.remove(path) instance = Instance.from_keyname(classname="clicsvtest", keyname="row4", expand=True) self.assertIsNotNone(instance) self.assertEqual(instance.fields["tags"].value, ["a", "b", "c"]) def test_csv_load_force_action_append_unions_values(self): """--force_action append reaches load_broker() and unions incoming values instead of replacing the field, same as the direct load_broker() path.""" load_broker( [{"classname": "clicsvtest", "keyname": "row5", "tags": ["a"]}], aaa={"perms": ["p_data_admin"]}, ) path = self._write_csv("keyname,schema,tags\nrow5,clicsvtest,b\n") try: output = StringIO() call_command( "cavaliba", "load", path, "--force_action", "append", verbosity=2, stdout=output, ) finally: os.remove(path) instance = Instance.from_keyname(classname="clicsvtest", keyname="row5", expand=True) self.assertEqual(instance.fields["tags"].value, ["a", "b"]) def test_csv_load_force_action_unappend_removes_values(self): """--force_action unappend reaches load_broker() and subtracts incoming values instead of replacing the field.""" load_broker( [{"classname": "clicsvtest", "keyname": "row6", "tags": ["a", "b"]}], aaa={"perms": ["p_data_admin"]}, ) path = self._write_csv("keyname,schema,tags\nrow6,clicsvtest,a\n") try: output = StringIO() call_command( "cavaliba", "load", path, "--force_action", "unappend", verbosity=2, stdout=output, ) finally: os.remove(path) instance = Instance.from_keyname(classname="clicsvtest", keyname="row6", expand=True) self.assertEqual(instance.fields["tags"].value, ["b"]) def test_csv_load_force_action_invalid_is_rejected(self): """An unsupported --force_action value is rejected before any load_broker call, mirroring the api.py _VALID_ACTIONS check.""" path = self._write_csv("keyname,schema,mystring\nrow7,clicsvtest,hello\n") try: output = StringIO() err = StringIO() call_command( "cavaliba", "load", path, "--force_action", "obliterate", verbosity=2, stdout=output, stderr=err, ) finally: os.remove(path) self.assertIn("ERR - invalid force_action", err.getvalue()) instance = Instance.from_keyname(classname="clicsvtest", keyname="row7", expand=False) self.assertIsNone(instance)