# (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"])
