# (c) cavaliba.com - tests / loader csv no pipeline

import os
import tempfile

from django.test import TestCase

import app_home.cache as cache
from app_data.loader import load_broker, load_file_csv
from tests.helper import add_schema


class TestLoaderCsvNoPipeline(TestCase):
    def setUp(self):
        cache.clear()

        add_schema(
            classname="csvtest",
            field_definition={
                "mystring": {"dataformat": "string", "displayname": "MyString"},
            },
        )
        add_schema(
            classname="csvtest2",
            field_definition={
                "mystring": {"dataformat": "string", "displayname": "MyString"},
            },
        )
        self.aaa = {"perms": ["p_data_admin"]}

    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_no_schema_column_returns_row_without_classname(self):
        """load_file_csv() no longer resolves a fallback classname itself - a row
        with no "schema" column comes back with no "classname" key at all."""

        path = self._write_csv("keyname,mystring\nrow1,hello\n")
        try:
            datalist, err = load_file_csv(filename=path)
        finally:
            os.remove(path)

        self.assertEqual(len(datalist), 1)
        self.assertNotIn("classname", datalist[0])
        self.assertEqual(datalist[0]["keyname"], "row1")

    def test_row_schema_column_is_renamed_to_classname(self):
        """A per-row "schema" column is renamed to "classname" (removed from the
        row dict)."""

        path = self._write_csv("keyname,schema,mystring\nrow1,csvtest2,hello\n")
        try:
            datalist, err = load_file_csv(filename=path)
        finally:
            os.remove(path)

        self.assertEqual(len(datalist), 1)
        self.assertEqual(datalist[0]["classname"], "csvtest2")
        self.assertNotIn("schema", datalist[0])

    def test_classname_column_is_not_a_resolution_source(self):
        """A "classname" column (as opposed to "schema") isn't specially handled -
        csv.DictReader already put its value straight into entry["classname"],
        load_file_csv() just never touches it."""

        path = self._write_csv("keyname,classname,mystring\nrow1,csvtest2,hello\n")
        try:
            datalist, err = load_file_csv(filename=path)
        finally:
            os.remove(path)

        self.assertEqual(datalist[0]["classname"], "csvtest2")

    def test_missing_keyname_row_still_returned(self):
        """A row with no "keyname" column value is no longer skipped by
        load_file_csv() itself - only load_broker()/load_instance() may reject it
        at write time."""

        path = self._write_csv("mystring\nhello\n")
        try:
            datalist, err = load_file_csv(filename=path)
        finally:
            os.remove(path)

        self.assertEqual(len(datalist), 1)
        self.assertNotIn("keyname", datalist[0])

    def test_load_broker_rejects_missing_classname(self):
        """Rows with no resolvable classname are rejected by load_broker(), not
        load_file_csv() - each row's rejection is collected as its own error."""

        path = self._write_csv("keyname,mystring\nrow1,hello\nrow2,world\n")
        try:
            datalist, err = load_file_csv(filename=path)
        finally:
            os.remove(path)

        result = load_broker(datalist=datalist, aaa=self.aaa)
        self.assertEqual(result["count_ko"], 2)
        self.assertIn("missing classname", result["errors"][0])
        self.assertIn("missing classname", result["errors"][1])

    def test_force_schema_supplies_missing_classname(self):
        """load_broker(force_schema=...) supplies a classname for rows that didn't
        have one - the replacement for the old load_file_csv(schema_name=)."""

        path = self._write_csv("keyname,mystring\nrow1,hello\n")
        try:
            datalist, err = load_file_csv(filename=path)
        finally:
            os.remove(path)

        result = load_broker(datalist=datalist, aaa=self.aaa, force_schema="csvtest")
        self.assertEqual(result["count_ok"], 1)