# (c) cavaliba.com - tests data export

import hashlib
import json
import os

from django.conf import settings
from django.test import TestCase, override_settings
from django.urls import reverse

import app_home.cache as cache
from app_data.data import Instance
from app_data.dataview import DataView
from app_data.revision import revision_add_raw
from tests import helper


class DataExportTest(TestCase):
    def setUp(self):
        cache.clear()
        helper.add_admin_user(login="unittest")
        helper.add_schema(
            classname="test1",
            field_definition={
                "mystring": {"dataformat": "string", "displayname": "MyString"},
            },
        )
        helper.add_instance(
            classname="test1",
            keyname="test1-01",
            fields={
                "mystring": "data for test1-01",
            },
        )
        helper.add_instance(
            classname="_dataview",
            keyname="test1_enumerate_inject",
            fields={
                "target_class": "test1",
                "content": "columns:\n  - keyname\n  - Widget:\n      from: my_enumerate_subfields__widget\n",
            },
        )
        helper.add_instance(
            classname="_dataview",
            keyname="test1_welcome",
            fields={
                "target_class": "test1",
                "content": "columns:\n  - keyname\n  - invalid:\n      from: non_existent_column\n",
            },
        )

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_data_export_csv(self):
        response = self.client.post(
            reverse("app_data:data_export"),
            {
                "classname": "test1",
                "page": "allpages",
                "dv": "___ALL___",
                "query": "",
                "format": "csv",
            },
        )

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "test1-01")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_data_export_yaml(self):
        response = self.client.post(
            reverse("app_data:data_export"),
            {
                "classname": "test1",
                "page": "allpages",
                "dv": "___ALL___",
                "query": "",
                "format": "yaml",
            },
        )

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "test1-01")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_data_export_json(self):
        response = self.client.post(
            reverse("app_data:data_export"),
            {
                "classname": "test1",
                "page": "allpages",
                "dv": "___ALL___",
                "query": "",
                "format": "json",
            },
        )

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "test1-01")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_data_export_yaml_dataview1(self):
        response = self.client.post(
            reverse("app_data:data_export"),
            {
                "classname": "test1",
                "page": "allpages",
                "dv": "test1_enumerate_inject",
                "query": "",
                "format": "yaml",
            },
        )

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "test1-01")
        # header/key is the real source field name, not the view's "Widget" label,
        # so the export stays importable
        self.assertContains(response, "my_enumerate_subfields__widget")
        self.assertNotContains(response, "Widget")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_data_export_yaml_dataview2(self):
        response = self.client.post(
            reverse("app_data:data_export"),
            {
                "classname": "test1",
                "page": "allpages",
                "dv": "test1_welcome",
                "query": "",
                "format": "yaml",
            },
        )

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "test1-01")
        # header/key is the real source field name, not the view's "invalid" label
        self.assertContains(response, "non_existent_column")
        self.assertNotContains(response, "invalid")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_data_export_json_dataview(self):
        response = self.client.post(
            reverse("app_data:data_export"),
            {
                "classname": "test1",
                "page": "allpages",
                "dv": "test1_welcome",
                "query": "",
                "format": "json",
            },
        )

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "test1-01")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_viewer_get(self):
        response = self.client.get(reverse("app_data:data_export"))

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Export Viewer")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_viewer_file_list_and_view(self):
        export_folder = str(settings.CAVALIBA_EXPORT_FOLDER)
        os.makedirs(export_folder, exist_ok=True)
        test_file = os.path.join(export_folder, "unittest.txt")
        test_content = "This is a test export file\n"
        with open(test_file, "w") as f:
            f.write(test_content)

        try:
            response = self.client.get(reverse("app_data:data_export"))
            self.assertEqual(response.status_code, 200)
            self.assertContains(response, "unittest.txt")

            filename_md5 = hashlib.md5(b"unittest.txt").hexdigest()

            download_response = self.client.get(
                reverse("app_data:data_export") + f"?dl={filename_md5}"
            )
            self.assertEqual(download_response.status_code, 200)
            self.assertIn("Content-Disposition", download_response)
            self.assertIn("attachment", download_response["Content-Disposition"])

            view_response = self.client.get(
                reverse("app_data:data_export") + f"?view={filename_md5}"
            )
            self.assertEqual(view_response.status_code, 200)
            self.assertContains(view_response, test_content.strip())
            self.assertContains(view_response, "unittest.txt")

        finally:
            if os.path.exists(test_file):
                os.remove(test_file)


class DataExportRefsRevTest(TestCase):
    def setUp(self):
        cache.clear()
        helper.add_admin_user(login="unittest")
        helper.add_user(login="testuser01")
        helper.add_schema(
            classname="test0",
            field_definition={
                "mystring": {"dataformat": "string", "displayname": "MyString"},
            },
        )
        helper.add_instance(
            classname="test0",
            keyname="test0-01",
            fields={
                "mystring": "data for test0-01",
            },
        )
        helper.add_schema(
            classname="test1",
            field_definition={
                "mystring": {"dataformat": "string", "displayname": "MyString"},
                "my_user": {"dataformat": "user", "displayname": "My User"},
                "my_schema_test0": {
                    "dataformat": "schema",
                    "dataformat_ext": "test0",
                    "displayname": "Schema Test0",
                },
            },
        )
        helper.add_instance(
            classname="test1",
            keyname="test1-01",
            fields={
                "mystring": "data for test1-01",
                "my_user": "testuser01",
                "my_schema_test0": "test0-01",
            },
        )

    def _post(self, extra=None):
        data = {
            "classname": "test1",
            "page": "allpages",
            "dv": "___ALL___",
            "query": "",
            "format": "yaml",
        }
        if extra:
            data.update(extra)
        return self.client.post(reverse("app_data:data_export"), data)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_yaml_refs_user(self):
        response = self._post({"refs": "user"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertIn("_refs", content)
        self.assertIn("testuser01", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_json_refs_user(self):
        response = self._post({"format": "json", "refs": "user"})
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.content)
        refs_found = any("_refs" in item for item in data)
        self.assertTrue(refs_found)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_yaml_refs_schema(self):
        response = self._post({"refs": "test0"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertIn("_refs", content)
        self.assertIn("test0", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_yaml_refs_empty_no_refs_key(self):
        response = self._post({"refs": ""})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertNotIn("_refs", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_yaml_refs_unknown_no_refs_key(self):
        response = self._post({"refs": "nonexistent_schema"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertNotIn("_refs", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_csv_refs_ignored(self):
        response = self._post({"format": "csv", "refs": "user"})
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "test1-01")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_yaml_rev_with_revisions(self):
        revision_add_raw(aaa=None, classname="test1", keyname="test1-01", action="edit")
        response = self._post({"rev": "3"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertIn("_revision", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_json_rev_with_revisions(self):
        revision_add_raw(aaa=None, classname="test1", keyname="test1-01", action="edit")
        response = self._post({"format": "json", "rev": "1"})
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.content)
        rev_found = any("_revision" in item for item in data)
        self.assertTrue(rev_found)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_yaml_rev_no_revisions(self):
        response = self._post({"rev": "5"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertNotIn("_revision", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_yaml_rev_invalid_value(self):
        response = self._post({"rev": "notanumber"})
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "test1-01")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_export_csv_rev_ignored(self):
        revision_add_raw(aaa=None, classname="test1", keyname="test1-01", action="edit")
        response = self._post({"format": "csv", "rev": "1"})
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "test1-01")


class TestGetDictForExportRevision(TestCase):
    def setUp(self):
        cache.clear()
        helper.add_schema(classname="revtest")
        helper.add_instance(classname="revtest", keyname="obj01")
        for i in range(15):
            revision_add_raw(classname="revtest", keyname="obj01", action="edit")

    def test_rev_none_no_revision_key(self):
        instance = Instance.from_keyname("revtest", "obj01")
        data = instance.get_dict_for_export(rev=None)
        self.assertNotIn("_revision", data)

    def test_rev_zero_no_revision_key(self):
        instance = Instance.from_keyname("revtest", "obj01")
        data = instance.get_dict_for_export(rev=0)
        self.assertNotIn("_revision", data)

    def test_rev_positive_returns_revision_key(self):
        instance = Instance.from_keyname("revtest", "obj01")
        data = instance.get_dict_for_export(rev=5)
        self.assertIn("_revision", data)

    def test_rev_limits_count(self):
        instance = Instance.from_keyname("revtest", "obj01")
        data = instance.get_dict_for_export(rev=5)
        self.assertEqual(len(data["_revision"]), 5)

    def test_rev_capped_at_available(self):
        instance = Instance.from_keyname("revtest", "obj01")
        data = instance.get_dict_for_export(rev=100)
        self.assertEqual(len(data["_revision"]), 15)

    def test_rev_entry_fields(self):
        instance = Instance.from_keyname("revtest", "obj01")
        data = instance.get_dict_for_export(rev=1)
        entry = data["_revision"][0]
        self.assertIn("login", entry)
        self.assertIn("date", entry)
        self.assertIn("action", entry)

    def test_rev_no_revisions_in_db(self):
        helper.add_instance(classname="revtest", keyname="obj02")
        instance = Instance.from_keyname("revtest", "obj02")
        data = instance.get_dict_for_export(rev=10)
        self.assertNotIn("_revision", data)


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

        # schema with a user field and a schema field
        helper.add_schema(
            classname="refstest",
            field_definition={
                "owner": {
                    "dataformat": "user",
                    "displayname": "Owner",
                    "cardinal_min": 0,
                    "cardinal_max": 1,
                },
                "linked": {
                    "dataformat": "schema",
                    "dataformat_ext": "refsother",
                    "displayname": "Linked",
                    "cardinal_min": 0,
                    "cardinal_max": 1,
                },
            },
        )
        helper.add_schema(classname="refsother")

        helper.add_instance(
            classname="refsother", keyname="other01", fields={"displayname": "Other 01"}
        )

        helper.add_instance(
            classname="refstest",
            keyname="obj01",
            fields={
                "owner": "testuser01",
                "linked": "other01",
            },
        )

        # user referenced by obj01, with extra fields for refs expansion
        user = helper.add_user(login="testuser01", displayname="Test User 01")
        user.set_field_value_single("email", "testuser01@example.com")
        user.set_field_value_single("mobile", "+33600000001")
        user.set_field_value_single("external_id", "EXT001")
        user.save(aaa={"username": "unittest"}, action="update")

    def _export(self, refs=None):
        instance = Instance.from_keyname("refstest", "obj01")
        return instance.get_dict_for_export(refs=refs)

    def test_refs_none_no_refs_key(self):
        data = self._export(refs=None)
        self.assertNotIn("_refs", data)

    def test_refs_empty_list_no_refs_key(self):
        data = self._export(refs=[])
        self.assertNotIn("_refs", data)

    def test_refs_unknown_type_no_refs_key(self):
        data = self._export(refs=["nonexistent"])
        self.assertNotIn("_refs", data)

    def test_refs_user_present(self):
        data = self._export(refs=["user"])
        self.assertIn("_refs", data)
        self.assertIn("user", data["_refs"])

    def test_refs_user_contains_expected_login(self):
        data = self._export(refs=["user"])
        logins = [u["keyname"] for u in data["_refs"]["user"]]
        self.assertIn("testuser01", logins)

    def test_refs_user_entry_fields(self):
        data = self._export(refs=["user"])
        entry = data["_refs"]["user"][0]
        self.assertIn("keyname", entry)
        self.assertIn("displayname", entry)

    def test_refs_user_extra_fields(self):
        data = self._export(refs=["user"])
        entry = next(u for u in data["_refs"]["user"] if u["keyname"] == "testuser01")
        self.assertEqual(entry.get("email"), "testuser01@example.com")
        self.assertEqual(entry.get("mobile"), "+33600000001")
        self.assertEqual(entry.get("external_id"), "EXT001")

    def test_refs_schema_present(self):
        data = self._export(refs=["refsother"])
        self.assertIn("_refs", data)
        self.assertIn("refsother", data["_refs"])

    def test_refs_schema_contains_expected_keyname(self):
        data = self._export(refs=["refsother"])
        keynames = [i["keyname"] for i in data["_refs"]["refsother"]]
        self.assertIn("other01", keynames)

    def test_refs_schema_entry_fields(self):
        data = self._export(refs=["refsother"])
        entry = data["_refs"]["refsother"][0]
        self.assertIn("keyname", entry)
        self.assertIn("classname", entry)
        self.assertIn("displayname", entry)

    def test_refs_multiple_types(self):
        data = self._export(refs=["user", "refsother"])
        self.assertIn("_refs", data)
        self.assertIn("user", data["_refs"])
        self.assertIn("refsother", data["_refs"])

    def test_refs_star_expands_to_schemas(self):
        data = self._export(refs=["*"])
        self.assertIn("_refs", data)
        self.assertIn("refsother", data["_refs"])

    def test_refs_star_with_user(self):
        data = self._export(refs=["*", "user"])
        self.assertIn("_refs", data)
        self.assertIn("user", data["_refs"])
        self.assertIn("refsother", data["_refs"])


class TestGetDictForExportFields(TestCase):
    def setUp(self):
        cache.clear()
        helper.add_schema(
            classname="fieldstest",
            field_definition={
                "alpha": {"dataformat": "string", "displayname": "Alpha"},
                "beta": {"dataformat": "string", "displayname": "Beta"},
                "charlie1": {"dataformat": "string", "displayname": "Charlie1"},
                "charlie2": {"dataformat": "string", "displayname": "Charlie2"},
            },
        )
        helper.add_instance(
            classname="fieldstest",
            keyname="obj01",
            fields={
                "alpha": "A",
                "beta": "B",
                "charlie1": "C1",
                "charlie2": "C2",
            },
        )

    def _export(self, fields=None):
        instance = Instance.from_keyname("fieldstest", "obj01")
        return instance.get_dict_for_export(fields=fields)

    def test_fields_none_all_present(self):
        data = self._export(fields=None)
        for f in ("alpha", "beta", "charlie1", "charlie2"):
            self.assertIn(f, data)

    def test_fields_star_all_present(self):
        data = self._export(fields="*")
        for f in ("alpha", "beta", "charlie1", "charlie2"):
            self.assertIn(f, data)

    def test_fields_empty_string_all_present(self):
        data = self._export(fields="")
        for f in ("alpha", "beta", "charlie1", "charlie2"):
            self.assertIn(f, data)

    def test_fields_whitelist_exact(self):
        data = self._export(fields="alpha,beta")
        self.assertIn("alpha", data)
        self.assertIn("beta", data)
        self.assertNotIn("charlie1", data)
        self.assertNotIn("charlie2", data)

    def test_fields_whitelist_wildcard(self):
        data = self._export(fields="charlie*")
        self.assertNotIn("alpha", data)
        self.assertNotIn("beta", data)
        self.assertIn("charlie1", data)
        self.assertIn("charlie2", data)

    def test_fields_blacklist_exact(self):
        data = self._export(fields="!alpha,beta")
        self.assertNotIn("alpha", data)
        self.assertNotIn("beta", data)
        self.assertIn("charlie1", data)
        self.assertIn("charlie2", data)

    def test_fields_blacklist_wildcard(self):
        data = self._export(fields="!charlie*")
        self.assertIn("alpha", data)
        self.assertIn("beta", data)
        self.assertNotIn("charlie1", data)
        self.assertNotIn("charlie2", data)

    def test_fields_exclude_all_keeps_builtin_attributes(self):
        data = self._export(fields="!*")
        self.assertNotIn("alpha", data)
        self.assertNotIn("beta", data)
        self.assertNotIn("charlie1", data)
        self.assertNotIn("charlie2", data)
        self.assertEqual(data["keyname"], "obj01")
        self.assertEqual(data["classname"], "fieldstest")
        self.assertNotIn("id", data)

    def test_fields_whitespace_tolerant(self):
        data = self._export(fields=" alpha , beta ")
        self.assertIn("alpha", data)
        self.assertIn("beta", data)
        self.assertNotIn("charlie1", data)


class TestGetCsvColumnsFields(TestCase):
    def setUp(self):
        cache.clear()
        helper.add_schema(
            classname="csvfieldstest",
            field_definition={
                "alpha": {"dataformat": "string", "displayname": "Alpha"},
                "charlie1": {"dataformat": "string", "displayname": "Charlie1"},
            },
        )
        helper.add_instance(
            classname="csvfieldstest",
            keyname="obj01",
            fields={"alpha": "A", "charlie1": "C1"},
        )

    def test_fields_none_all_columns(self):
        instance = Instance.from_keyname("csvfieldstest", "obj01")
        columns = instance.get_csv_columns(fields=None)
        self.assertIn("alpha", columns)
        self.assertIn("charlie1", columns)

    def test_fields_whitelist(self):
        instance = Instance.from_keyname("csvfieldstest", "obj01")
        columns = instance.get_csv_columns(fields="alpha")
        self.assertIn("alpha", columns)
        self.assertNotIn("charlie1", columns)

    def test_fields_blacklist_keeps_only_keyname(self):
        # keyname is the only fixed column - schema/is_enabled/displayname are
        # all opt-out via the fields filter, same as regular fields
        instance = Instance.from_keyname("csvfieldstest", "obj01")
        columns = instance.get_csv_columns(fields="!*")
        self.assertNotIn("alpha", columns)
        self.assertNotIn("charlie1", columns)
        self.assertNotIn("classname", columns)
        self.assertNotIn("schema", columns)
        self.assertIn("keyname", columns)
        self.assertNotIn("is_enabled", columns)
        self.assertNotIn("displayname", columns)

    def test_fields_none_includes_displayname(self):
        # no filter provided means "*" (all fields, including displayname)
        instance = Instance.from_keyname("csvfieldstest", "obj01")
        columns = instance.get_csv_columns(fields=None)
        self.assertIn("displayname", columns)
        self.assertIn("schema", columns)
        self.assertNotIn("classname", columns)
        self.assertIn("keyname", columns)
        self.assertIn("is_enabled", columns)

    def test_fields_star_includes_displayname(self):
        instance = Instance.from_keyname("csvfieldstest", "obj01")
        columns = instance.get_csv_columns(fields="*")
        self.assertIn("displayname", columns)

    def test_fields_explicit_whitelist_includes_displayname(self):
        instance = Instance.from_keyname("csvfieldstest", "obj01")
        columns = instance.get_csv_columns(fields="displayname")
        self.assertIn("displayname", columns)
        self.assertNotIn("alpha", columns)


class DataExportFieldsViewTest(TestCase):
    def setUp(self):
        cache.clear()
        helper.add_admin_user(login="unittest")
        helper.add_schema(
            classname="test2",
            field_definition={
                "keep_me": {"dataformat": "string", "displayname": "Keep Me"},
                "drop_me": {"dataformat": "string", "displayname": "Drop Me"},
            },
        )
        helper.add_instance(
            classname="test2",
            keyname="test2-01",
            fields={"keep_me": "keepvalue", "drop_me": "dropvalue"},
        )
        helper.add_instance(
            classname="_dataview",
            keyname="test2_view",
            fields={
                "target_class": "test2",
                "content": "columns:\n  - keyname\n  - drop_me\n",
            },
        )

    def _post(self, extra=None):
        data = {
            "classname": "test2",
            "page": "allpages",
            "dv": "___ALL___",
            "query": "",
            "format": "yaml",
        }
        if extra:
            data.update(extra)
        return self.client.post(reverse("app_data:data_export"), data)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_fields_whitelist_excludes_other_field(self):
        response = self._post({"fields": "keep_me"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertIn("keepvalue", content)
        self.assertNotIn("dropvalue", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_fields_blacklist_excludes_field(self):
        response = self._post({"fields": "!drop_me"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertIn("keepvalue", content)
        self.assertNotIn("dropvalue", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_fields_default_includes_all(self):
        response = self._post()
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertIn("keepvalue", content)
        self.assertIn("dropvalue", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_fields_csv_whitelist(self):
        response = self._post({"format": "csv", "fields": "keep_me"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertIn("keepvalue", content)
        self.assertNotIn("dropvalue", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_fields_blacklist_applies_with_named_dataview(self):
        # even though the dataview explicitly lists "drop_me" as a column,
        # the fields filter still drops it
        response = self._post({"dv": "test2_view", "fields": "!drop_me"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertNotIn("dropvalue", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_fields_whitelist_applies_with_named_dataview(self):
        response = self._post({"dv": "test2_view", "fields": "keyname"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertNotIn("dropvalue", content)
        self.assertIn("test2-01", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_fields_default_keeps_dataview_columns(self):
        response = self._post({"dv": "test2_view"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertIn("dropvalue", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_fields_blacklist_applies_to_renamed_dataview_column(self):
        # dataview column is a dict-form rename ("MyDrop" label, "from: drop_me"
        # source) - the filter must match the source field, not the label
        helper.add_instance(
            classname="_dataview",
            keyname="test2_view_renamed",
            fields={
                "target_class": "test2",
                "content": "columns:\n  - keyname\n  - MyDrop:\n      from: drop_me\n",
            },
        )
        response = self._post({"dv": "test2_view_renamed", "fields": "!drop_me"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertNotIn("dropvalue", content)
        self.assertNotIn("MyDrop", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_fields_keyname_always_included_in_dataview(self):
        response = self._post({"dv": "test2_view", "fields": "!*"})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertNotIn("dropvalue", content)
        self.assertIn("test2-01", content)


class TestDataViewFieldsFilter(TestCase):
    def setUp(self):
        cache.clear()
        helper.add_schema(
            classname="dvfieldstest",
            field_definition={
                "keep_me": {"dataformat": "string", "displayname": "Keep Me"},
                "drop_me": {"dataformat": "string", "displayname": "Drop Me"},
            },
        )
        helper.add_instance(
            classname="dvfieldstest",
            keyname="obj01",
            fields={"keep_me": "keepvalue", "drop_me": "dropvalue"},
        )
        helper.add_instance(
            classname="_dataview",
            keyname="dvfieldstest_plain",
            fields={
                "target_class": "dvfieldstest",
                "content": "columns:\n  - keyname\n  - keep_me\n  - drop_me\n",
            },
        )
        helper.add_instance(
            classname="_dataview",
            keyname="dvfieldstest_renamed",
            fields={
                "target_class": "dvfieldstest",
                "content": "columns:\n  - keyname\n  - MyDrop:\n      from: drop_me\n",
            },
        )
        helper.add_instance(
            classname="_dataview",
            keyname="dvfieldstest_classname",
            fields={
                "target_class": "dvfieldstest",
                "content": (
                    "columns:\n  - keyname\n  - classname\n  - schema:\n      from: classname\n"
                ),
            },
        )
        helper.add_instance(
            classname="_dataview",
            keyname="dvfieldstest_no_identity",
            fields={
                "target_class": "dvfieldstest",
                "content": "columns:\n  - keep_me\n",
            },
        )
        helper.add_instance(
            classname="_dataview",
            keyname="dvfieldstest_empty",
            fields={
                "target_class": "dvfieldstest",
                "content": "columns: []\n",
            },
        )

    def test_get_columns_none_all_columns(self):
        # classname isn't in this view's own columns, so it's guaranteed/prepended
        dataview = DataView(keyname="dvfieldstest_plain")
        columns = dataview.get_columns(fields=None)
        self.assertEqual(columns, ["classname", "keyname", "keep_me", "drop_me"])

    def test_get_columns_whitelist(self):
        dataview = DataView(keyname="dvfieldstest_plain")
        columns = dataview.get_columns(fields="keep_me")
        self.assertEqual(columns, ["classname", "keyname", "keep_me"])

    def test_get_columns_blacklist(self):
        dataview = DataView(keyname="dvfieldstest_plain")
        columns = dataview.get_columns(fields="!drop_me")
        self.assertEqual(columns, ["classname", "keyname", "keep_me"])

    def test_get_columns_blacklist_matches_renamed_column_source(self):
        dataview = DataView(keyname="dvfieldstest_renamed")
        columns = dataview.get_columns(fields="!drop_me")
        self.assertEqual(columns, ["classname", "keyname"])

    def test_get_columns_whitelist_keeps_keyname(self):
        dataview = DataView(keyname="dvfieldstest_plain")
        columns = dataview.get_columns(fields="!*")
        self.assertEqual(columns, ["classname", "keyname"])

    def test_get_columns_guarantees_classname_and_keyname_by_source_not_label(self):
        # dvfieldstest_classname explicitly defines both (keyname plain, classname
        # via plain + a "schema" rename, deduped down to one) - already covered,
        # so no forced prepend
        dataview = DataView(keyname="dvfieldstest_classname")
        columns = dataview.get_columns()
        self.assertEqual(columns, ["keyname", "classname"])

    def test_get_columns_renamed_column_uses_source_name(self):
        # the real-world case: a view renames "keyname" to a human label (here
        # drop_me -> "MyDrop") - the export column must be the real field name
        dataview = DataView(keyname="dvfieldstest_renamed")
        columns = dataview.get_columns()
        self.assertEqual(columns, ["classname", "keyname", "drop_me"])
        self.assertNotIn("MyDrop", columns)

    def test_filter_renamed_column_value_keyed_by_source_name(self):
        instance = Instance.from_keyname("dvfieldstest", "obj01")
        dataview = DataView(keyname="dvfieldstest_renamed")
        columns = dataview.get_columns()
        values = dataview.filter(instance=instance, mode="csv")
        result = dict(zip(columns, values))
        self.assertEqual(result["drop_me"], "dropvalue")
        self.assertNotIn("MyDrop", result)

    def test_get_columns_view_with_neither_identity_column(self):
        # a view listing only a regular field still gets classname/keyname
        # prepended, so the export stays re-importable via cavaliba load
        dataview = DataView(keyname="dvfieldstest_no_identity")
        columns = dataview.get_columns()
        self.assertEqual(columns, ["classname", "keyname", "keep_me"])

    def test_get_columns_empty_view_still_guarantees_identity(self):
        dataview = DataView(keyname="dvfieldstest_empty")
        columns = dataview.get_columns()
        self.assertEqual(columns, ["classname", "keyname"])

    def test_filter_view_with_neither_identity_column(self):
        instance = Instance.from_keyname("dvfieldstest", "obj01")
        dataview = DataView(keyname="dvfieldstest_no_identity")
        values = dataview.filter(instance=instance, mode="csv")
        columns = dataview.get_columns()
        result = dict(zip(columns, values))
        self.assertEqual(result["classname"], "dvfieldstest")
        self.assertEqual(result["keyname"], "obj01")
        self.assertEqual(result["keep_me"], "keepvalue")

    def test_filter_empty_view_still_guarantees_identity(self):
        instance = Instance.from_keyname("dvfieldstest", "obj01")
        dataview = DataView(keyname="dvfieldstest_empty")
        values = dataview.filter(instance=instance, mode="csv")
        self.assertEqual(values, ["dvfieldstest", "obj01"])

    def test_filter_and_get_columns_stay_aligned_with_action_and_no_identity(self):
        instance = Instance.from_keyname("dvfieldstest", "obj01")
        dataview = DataView(keyname="dvfieldstest_no_identity")
        columns = dataview.get_columns(action="noop")
        values = dataview.filter(instance=instance, mode="csv", action="noop")
        self.assertEqual(columns, ["_action", "classname", "keyname", "keep_me"])
        self.assertEqual(values, ["noop", "dvfieldstest", "obj01", "keepvalue"])

    def test_filter_blacklist_drops_value(self):
        instance = Instance.from_keyname("dvfieldstest", "obj01")
        dataview = DataView(keyname="dvfieldstest_plain")
        values = dataview.filter(instance=instance, mode="value", fields="!drop_me")
        # classname, keyname, keep_me (drop_me blacklisted)
        self.assertEqual(len(values), 3)

    def test_filter_stays_aligned_with_get_columns(self):
        instance = Instance.from_keyname("dvfieldstest", "obj01")
        dataview = DataView(keyname="dvfieldstest_plain")
        columns = dataview.get_columns(fields="!drop_me")
        values = dataview.filter(instance=instance, mode="csv", fields="!drop_me")
        self.assertEqual(len(columns), len(values))
        self.assertEqual(dict(zip(columns, values))["keep_me"], "keepvalue")

    def test_classname_plain_column_resolves_value(self):
        instance = Instance.from_keyname("dvfieldstest", "obj01")
        dataview = DataView(keyname="dvfieldstest_classname")
        columns = dataview.get_columns()
        values = dataview.filter(instance=instance, mode="csv")
        self.assertEqual(dict(zip(columns, values))["classname"], "dvfieldstest")

    def test_classname_duplicate_source_deduplicated(self):
        # view has both a plain "classname" column and a "schema" rename of
        # "classname" - the second (same source) is dropped, not double-exported
        instance = Instance.from_keyname("dvfieldstest", "obj01")
        dataview = DataView(keyname="dvfieldstest_classname")
        columns = dataview.get_columns()
        values = dataview.filter(instance=instance, mode="csv")
        self.assertEqual(columns, ["keyname", "classname"])
        self.assertEqual(len(columns), len(values))
        self.assertNotIn("schema", columns)

    def test_classname_survives_blacklist(self):
        dataview = DataView(keyname="dvfieldstest_classname")
        columns = dataview.get_columns(fields="!*")
        self.assertEqual(columns, ["keyname", "classname"])


class DataExportActionNoopTest(TestCase):
    def setUp(self):
        cache.clear()
        helper.add_admin_user(login="unittest")
        helper.add_schema(
            classname="noopmount",
            field_definition={
                "mystring": {"dataformat": "string", "displayname": "MyString"},
            },
        )
        helper.add_instance(
            classname="noopmount",
            keyname="noop-01",
            fields={"mystring": "value1"},
        )

    def _post(self, extra=None):
        data = {
            "classname": "noopmount",
            "page": "allpages",
            "dv": "___ALL___",
            "query": "",
            "format": "yaml",
        }
        if extra:
            data.update(extra)
        return self.client.post(reverse("app_data:data_export"), data)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_yaml_includes_action_noop(self):
        response = self._post()
        content = response.content.decode()
        self.assertIn("_action: noop", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_json_includes_action_noop(self):
        response = self._post({"format": "json"})
        data = json.loads(response.content)
        self.assertTrue(all(item.get("_action") == "noop" for item in data))

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_csv_has_no_action_column(self):
        # unlike YAML/JSON, CSV never gets a safe-by-default "_action" column
        response = self._post({"format": "csv"})
        content = response.content.decode()
        header = content.strip().splitlines()[0].split(",")
        self.assertNotIn("_action", header)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_csv_has_no_action_column_with_named_dataview(self):
        helper.add_instance(
            classname="_dataview",
            keyname="noopmount_view",
            fields={
                "target_class": "noopmount",
                "content": "columns:\n  - keyname\n  - mystring\n",
            },
        )
        response = self._post({"format": "csv", "dv": "noopmount_view"})
        content = response.content.decode()
        header = content.strip().splitlines()[0].split(",")
        self.assertNotIn("_action", header)


class DataExportCsvOptionsTest(TestCase):
    def setUp(self):
        cache.clear()
        helper.add_admin_user(login="unittest")
        helper.add_schema(
            classname="csvopttest",
            field_definition={
                "mystring": {"dataformat": "string", "displayname": "MyString"},
            },
        )
        helper.add_instance(
            classname="csvopttest",
            keyname="csvopt-01",
            fields={"mystring": "café"},
        )

    def _post(self, extra=None, follow=False):
        data = {
            "classname": "csvopttest",
            "page": "allpages",
            "dv": "___ALL___",
            "query": "",
            "format": "csv",
        }
        if extra:
            data.update(extra)
        return self.client.post(reverse("app_data:data_export"), data, follow=follow)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_default_delimiter_is_comma_family(self):
        response = self._post()
        self.assertEqual(response.status_code, 200)
        content = response.content.decode()
        self.assertIn("csvopt-01", content)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_custom_delimiter_semicolon(self):
        response = self._post({"delimiter": ";"})
        self.assertEqual(response.status_code, 200)
        header = response.content.decode().splitlines()[0]
        self.assertIn(";", header)
        self.assertNotIn(",", header)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_custom_delimiter_pipe(self):
        response = self._post({"delimiter": "|"})
        self.assertEqual(response.status_code, 200)
        header = response.content.decode().splitlines()[0]
        self.assertIn("|", header)

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_invalid_delimiter_rejected(self):
        response = self._post({"delimiter": "::"}, follow=True)
        self.assertContains(response, "single character")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_default_encoding_is_utf8(self):
        response = self._post()
        self.assertEqual(response.status_code, 200)
        self.assertIn("utf-8", response["Content-Type"])
        self.assertIn("café", response.content.decode("utf-8"))

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_custom_encoding_latin1(self):
        response = self._post({"encoding": "iso-8859-1"})
        self.assertEqual(response.status_code, 200)
        self.assertIn("iso-8859-1", response["Content-Type"])
        self.assertIn("café", response.content.decode("iso-8859-1"))

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_unknown_encoding_rejected(self):
        response = self._post({"encoding": "not-a-real-encoding"}, follow=True)
        self.assertContains(response, "Unknown CSV encoding")

    @override_settings(CAVALIBA_AUTH_MODE="unittest")
    def test_encoding_cannot_represent_data_rejected(self):
        response = self._post({"encoding": "ascii"}, follow=True)
        self.assertContains(response, "cannot be encoded")
