diff --git a/docs/openedx_tagging/decisions/0014-archived-tags-on-create-import-export.rst b/docs/openedx_tagging/decisions/0014-archived-tags-on-create-import-export.rst new file mode 100644 index 000000000..a9ed78569 --- /dev/null +++ b/docs/openedx_tagging/decisions/0014-archived-tags-on-create-import-export.rst @@ -0,0 +1,167 @@ +.. _openedx-tagging-adr-0014: + +14. Archived tags on create, import, and export +================================================ + +Status +------ + +Proposed + +Several decisions below are marked **(awaiting confirmation)**. They need an answer from the +owner named in each one before this ADR can be accepted. + +Context +------- + +Competency-based education (CBE) records learner mastery against tags. Deleting a tag would +destroy the history that points at it, so the approved design on `#655`_ archives a tag that +learner mastery depends on instead of deleting it (see also :ref:`openedx_learning ADR 0003 +`, decision 3). +`#776`_ adds the ``Tag.archived`` flag, `#778`_ hides archived tags from the read paths and the +export through ``TagQuerySet.unarchived()``, and `#780`_ turns a tag delete into an archive when +something depends on the tag. + +An archived tag stays in the database and keeps its ``value`` and ``external_id``. The +per-taxonomy ``unique_together`` constraints on ``(taxonomy, value)`` and +``(taxonomy, external_id)`` have no exception for archived rows, so both values stay taken. +No API function or REST endpoint can restore an archived tag. An operator can still clear ``archived`` +in the Django admin, as "Known limitations" describes. + +This ADR records how the paths that write tags into a taxonomy treat an archived tag: creating a +tag (``Taxonomy.add_tag``, ``add_tag_to_taxonomy`` and the REST create endpoint), importing a file +(``TagImportPlan`` through ``import_tags``), and exporting a taxonomy. It covers ``Tag`` only. +Archived taxonomies and archived tag associations (``ObjectTag``) follow `#778`_ and `#779`_. + +Decision +-------- + +An archived tag is no longer part of its taxonomy +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +No path in this ADR creates, updates, renames, re-parents or revives an archived tag. The tag +still occupies its value and its ``external_id``, and those values stay reserved for as long as +the tag exists. + +The archived subtree +~~~~~~~~~~~~~~~~~~~~ + +The **archived subtree** of a taxonomy is every archived tag together with every tag under one. +``TagQuerySet.in_archived_subtree()`` returns it. Because tree depth is capped by +``TAXONOMY_MAX_DEPTH``, the query checks the tag and a fixed number of ancestors, so its size does +not grow with the number of archived tags. + +Create, import, and the replace sweep treat every tag in the archived subtree as outside the +taxonomy. The export from `#778`_ already leaves the whole subtree out. + +**(Awaiting confirmation from the #778 owner.)** Whether an unarchived tag under an archived tag +can exist at all is open. If archiving always covers the whole subtree, as `#780`_'s rules produce, +the subtree rule above only guards against direct database edits. If such a tag can exist, the +listings and ``update_tag`` from `#778`_ still treat it as live, while the paths in this ADR do +not. + +Create +~~~~~~ + +Adding a tag whose value or ``external_id`` belongs to an archived tag raises ``ValueError`` with a +message that names the archived tag as the cause and asks for a different value. A value held by an +unarchived tag keeps the existing "already exists" message. Adding a tag under a tag in the +archived subtree is refused. Values are stripped before they are compared, as ``Tag.clean`` strips +them before saving. + +Import +~~~~~~ + +A row whose ``value``, ``id``, ``previous_id`` or ``parent_id`` matches a tag in the archived +subtree is an import plan error. When any row has this error, the plan builds no actions, so the +import changes nothing. The error is reported once per row. + +The match runs in the database, under the column collation, because the import actions find the +tags they change with database lookups. A comparison in Python would disagree with the database: +on MySQL the collation also ignores accents and trailing spaces. + +A replace import never deletes a tag in the archived subtree, because the export leaves that +subtree out and its absence from the file says nothing about it. These checks hold for the tags +that are archived when the plan is built. + +Export +~~~~~~ + +An export leaves out the archived subtree (`#778`_). Importing that export into an empty taxonomy, +or re-importing it into its own taxonomy with ``replace=True``, succeeds and leaves the archived +subtree as it was. + +Known limitations +~~~~~~~~~~~~~~~~~ + +- **(Awaiting confirmation from the #780 owner.)** A replace import that deletes the parent of an + archived tag moves the archived tag to the root. This rewrites the parent of an archived record. + The recommended rule is to archive such a parent instead, the way `#780`_ archives the ancestors + of a depended-upon tag. Moving the tag to the root is the interim behavior until the import's + deletes follow `#780`_. +- **(Awaiting confirmation from the #780 owner.)** A replace import deletes through its own + ``DeleteTag`` action, which `#780`_ does not cover. Until the import's deletes go through the + archive-or-delete decision of `#780`_, a replace import can hard-delete a tag that learner mastery + depends on. The recommended shape is one decision for the whole delete set, made inside the + import's ``execute`` transaction. +- Course import reuses ``TagImportPlan`` (:ref:`ADR 0011 `). + A course whose tags match an archived tag on the target instance therefore fails its criteria + import with this ADR's plan error, although the course author cannot change the taxonomy. ADR + 0011 should be amended for this case separately. +- The error quotes the administrator's input, not the archived tag that holds the value. An + administrator cannot see that tag, for example the archived "Cafe" that refuses "Café" on MySQL. +- ``Taxonomy.update_tag`` renaming a tag to an archived value raises ``IntegrityError`` rather than + a message that names the archived tag. The REST rename path reports it through the serializer. +- The Django admin edit page for ``Tag`` does not restrict its fields, so an operator can change + ``archived`` and an archived tag's value there. Such changes are outside the rules in this ADR. + +Future considerations +~~~~~~~~~~~~~~~~~~~~~ + +**(Awaiting confirmation from product.)** The approved thread on `#655`_ returns an error "for now" +and expects that adding a tag with an archived tag's value may later un-archive that tag instead. +The reserved values above describe the current state, not a permanent rule. A restore path would +also let the messages suggest restoring the tag instead of choosing a different value. + +Rejected Alternatives +--------------------- + +Un-archive the tag when its value is created or imported again +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Rejected for now because no restore path has been designed: the approved design on `#655`_ chose an +error, and silently reviving a tag would reattach learner history to a tag an administrator chose to +remove. + +Release the archived tag's value and ``external_id`` +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Making the unique constraints conditional on ``archived`` would let a new tag take the value. +Rejected because two tags in one taxonomy would then share a value and an ``external_id``, which +breaks import and export, both keyed on them, and any later restore. + +Find the archived subtree by lineage prefix +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +One ``lineage__startswith`` condition per archived tag selects the same rows, but the query grows +with every archived tag. SQLite refuses an expression tree deeper than 1,000, so a taxonomy with +about 1,000 archived tags could no longer be imported at all. + +Delete the archived subtree on a replace import +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +Rejected because the export omits the subtree, so every round trip of a taxonomy through export and +replace import would destroy the history that archiving exists to keep. + +Changelog +--------- + +2026-11-06: + +* Proposed. + +.. _#655: https://github.com/openedx/openedx-core/issues/655 +.. _#776: https://github.com/openedx/openedx-core/issues/776 +.. _#778: https://github.com/openedx/openedx-core/issues/778 +.. _#779: https://github.com/openedx/openedx-core/issues/779 +.. _#780: https://github.com/openedx/openedx-core/issues/780 diff --git a/src/openedx_tagging/api.py b/src/openedx_tagging/api.py index 005600d63..1e6d71190 100644 --- a/src/openedx_tagging/api.py +++ b/src/openedx_tagging/api.py @@ -82,28 +82,28 @@ def create_taxonomy( # pylint: disable=too-many-positional-arguments def get_taxonomy(taxonomy_id: int) -> Taxonomy | None: """ - Returns the Taxonomy which has the given ID, or None if not found. + Returns the Taxonomy which has the given ID, or None if not found or archived. """ - return Taxonomy.objects.filter(pk=taxonomy_id).first() + return Taxonomy.objects.unarchived().filter(pk=taxonomy_id).first() def get_taxonomy_by_export_id(taxonomy_export_id: str) -> Taxonomy | None: """ - Returns the Taxonomy which has the given export ID, or None if not found. + Returns the Taxonomy which has the given export ID, or None if not found or archived. """ - return Taxonomy.objects.filter(export_id=taxonomy_export_id).first() + return Taxonomy.objects.unarchived().filter(export_id=taxonomy_export_id).first() def get_taxonomies(enabled=True) -> QuerySet[Taxonomy]: """ - Returns a queryset containing the enabled taxonomies, sorted by name. + Returns a queryset containing the enabled, unarchived taxonomies, sorted by name. We return a QuerySet here for ease of use with Django Rest Framework and other query-based use cases. If you want the disabled taxonomies, pass enabled=False. If you want all taxonomies (both enabled and disabled), pass enabled=None. """ - queryset = Taxonomy.objects.order_by("name", "id") + queryset = Taxonomy.objects.unarchived().order_by("name", "id") if enabled is None: return queryset.all() return queryset.filter(enabled=enabled) @@ -111,7 +111,7 @@ def get_taxonomies(enabled=True) -> QuerySet[Taxonomy]: def get_tags(taxonomy: Taxonomy) -> TagDataQuerySet: """ - Returns a QuerySet of all the tags in the given taxonomy. + Returns a QuerySet of all the unarchived tags in the given taxonomy. Note that if the taxonomy is dynamic or free-text, only tags that have already been applied to some object will be returned. @@ -146,7 +146,7 @@ def search_tags( # Fetch tags that the object already has to exclude them from the result. # Note: this adds a fair bit of complexity. In the future, maybe we can just do this filtering on the frontend? excluded_values = list( - taxonomy.objecttag_set.filter(object_id=exclude_object_id).values_list( + taxonomy.objecttag_set.unarchived().filter(object_id=exclude_object_id).values_list( "_value", flat=True ) ) @@ -162,7 +162,9 @@ def get_children_tags( parent_tag_value: str, ) -> TagDataQuerySet: """ - Returns a QuerySet of children tags for the given parent tag. + Returns a QuerySet of unarchived children tags for the given parent tag. + + Raises Tag.DoesNotExist if the parent tag does not exist, or Tag.IsArchived if it is archived. Note that if the taxonomy allows free-text tags, then the returned list will be empty. """ @@ -196,11 +198,14 @@ def get_object_tags( """ Returns a Queryset of object tags for a given object. + Archived object tags are excluded, as are those whose tag or taxonomy is archived. + Pass taxonomy_id to limit the returned object_tags to a specific taxonomy. """ filters = {"taxonomy_id": taxonomy_id} if taxonomy_id else {} base_qs = ( ObjectTag.objects + .unarchived() .filter(object_id=object_id, **filters) .exclude(taxonomy__enabled=False) # Exclude if the whole taxonomy is disabled ) @@ -234,14 +239,15 @@ def get_object_tag_counts(object_id_pattern: str, count_implicit=False) -> dict[ "course-v1:foo+bar+baz@*", or a list of "comma,separated,IDs", return a dict of matching object IDs and how many tags each object has. - Deleted tags and disabled taxonomies are excluded from the counts, even if + Deleted tags, disabled taxonomies, and archived object tags (including those + whose tag or taxonomy is archived) are excluded from the counts, even if ObjectTag data about them is present. count_implicit: if True, this means to count all ancestor tags (implicit tags) of the explict tags that are associated with the object. """ # Note: in the future we may add an option to exclude system taxonomies from the count. - qs: Any = ObjectTag.objects + qs: Any = ObjectTag.objects.unarchived() if object_id_pattern.endswith("*"): qs = qs.filter(object_id__startswith=object_id_pattern[0:len(object_id_pattern) - 1]) elif "*" in object_id_pattern: @@ -449,6 +455,9 @@ def add_tag_to_taxonomy( Tag is returned. If `external_id` is omitted, one is generated from the tag's name. + + Raises ``ValueError`` if the value or ``external_id`` is held by another tag, + archived or not, or if the parent is archived or under an archived tag. """ new_tag = taxonomy.add_tag(tag, parent_tag_value, external_id) @@ -549,13 +558,15 @@ def add_usage_counts(taxonomy: Taxonomy, tag_data: TagDataQuerySet) -> TagDataQu parent tag, "Natural Science" is only applied to it once, because no tag can be applied to the same object twice. + Archived object tags, and those whose tag or taxonomy is archived, are not counted. + For performance reasons, we call this function with the list result of the QuerySet so we can then add the counts in-memory rather than annotate to a QuerySet which would require a very expensive annotation to join the in-memory data to the original QuerySet. """ - object_tags = taxonomy.objecttag_set.values_list("object_id", "tag__lineage") + object_tags = taxonomy.objecttag_set.unarchived().values_list("object_id", "tag__lineage") tag_counts: Counter[str] = Counter() object_tag_lineage_seen: defaultdict[str, set] = defaultdict(set) diff --git a/src/openedx_tagging/import_export/api.py b/src/openedx_tagging/import_export/api.py index c9ac1434b..09b7f2756 100644 --- a/src/openedx_tagging/import_export/api.py +++ b/src/openedx_tagging/import_export/api.py @@ -71,10 +71,14 @@ def import_tags( of each import step and the current status. There can only be one task in progress at a time per taxonomy - Set `replace` to True to delete all not readed Tag of the given taxonomy. + Set `replace` to True to delete every tag of the given taxonomy that the file does not mention. Ex. Given a taxonomy with `tag_1`, `tag_2` and `tag_3`. If there is only `tag_1` in the file (regardless of action), then `tag_2` and `tag_3` will be deleted - if `replace=True` + if `replace=True`. A tag in an archived subtree when the plan is built is not deleted this way. + An archived tag whose parent is deleted is moved to the root. + + A file row whose value, id, previous_id or parent_id belongs to a tag in an archived + subtree is reported as an error, and nothing is imported. Set `plan_only` to True to only generate the actions and not execute them. """ diff --git a/src/openedx_tagging/import_export/exceptions.py b/src/openedx_tagging/import_export/exceptions.py index d248f0006..3da8deb1b 100644 --- a/src/openedx_tagging/import_export/exceptions.py +++ b/src/openedx_tagging/import_export/exceptions.py @@ -80,6 +80,24 @@ def __init__(self, tag_id: str, plain_row: int, rename_row: int, **kargs): ).format(tag_id=tag_id, plain_row=plain_row, rename_row=rename_row) +class ArchivedTagCollisionError(TagImportError): + """ + Exception raised when a row's value, id, previous_id or parent_id matches an archived tag or a tag under one. + + ``row`` is the row's 1-based position among the file's data rows. + """ + + def __init__(self, row: int, field: str, value: str, under_archived: bool = False, **kargs): + super().__init__(**kargs) + if under_archived: + message = _("Row {row}'s {field} ({value}) belongs to a tag under an archived tag in this taxonomy.") + else: + message = _("Row {row}'s {field} ({value}) belongs to an archived tag in this taxonomy.") + # previous_id names the one tag the row means to rename, so no other value would do. + advice = _("An import cannot change it.") if field == "previous_id" else _("Use a different {field}.") + self.message = f"{message} {advice}".format(row=row, field=field, value=value) + + class ParentCycleError(TagImportError): """ Exception raised when a row's parent_id would make its tag its own ancestor. diff --git a/src/openedx_tagging/import_export/import_plan.py b/src/openedx_tagging/import_export/import_plan.py index e7c4c74c3..c36291be1 100644 --- a/src/openedx_tagging/import_export/import_plan.py +++ b/src/openedx_tagging/import_export/import_plan.py @@ -6,7 +6,8 @@ from typing import NamedTuple from attrs import define -from django.db import transaction +from django.db import models, transaction +from django.db.models import Q from ..models import Tag, TagImportTask, Taxonomy from ..models.base import TAXONOMY_MAX_DEPTH @@ -21,6 +22,7 @@ available_actions, ) from .exceptions import ( + ArchivedTagCollisionError, DuplicateFinalIdError, ParentCycleError, StaleIdTargetedByPlainRowError, @@ -94,6 +96,10 @@ class TagImportPlan: actions_dict: dict taxonomy: Taxonomy + # A row adds up to four query parameters, so a chunk stays under the 999 that + # Django's SQLite backend assumes. + IN_CHUNK_SIZE = 200 + def __init__(self, taxonomy: Taxonomy): self.actions = [] self.errors = [] @@ -245,6 +251,67 @@ def _validate_no_plain_row_targets_a_vacated_id(self, tags: list[TagItem]) -> No if rename_row is not None: self.errors.append(StaleIdTargetedByPlainRowError(tag.id, position, rename_row)) + @staticmethod + def _row_keys(tag: TagItem) -> list[tuple[str, str, str]]: + """ + Return the (field, value, lookup) triples a row can match an existing tag on. + + Values are stripped, as `Tag.clean` strips them before saving. + """ + keys = [ + ("value", tag.value, "value"), + ("id", tag.id, "external_id"), + ("previous_id", tag.previous_id, "external_id"), + ("parent_id", tag.parent_id, "external_id"), + ] + return [(field, value.strip(), lookup) for field, value, lookup in keys if value and value.strip()] + + def _chunk_matches_archived_subtree(self, chunk: list[TagItem], archived_subtree: models.QuerySet[Tag]) -> bool: + """ + Return whether any row in the chunk matches a tag in `archived_subtree`, in one query. + """ + values: dict[str, list[str]] = {"value": [], "external_id": []} + for tag in chunk: + for _, value, lookup in self._row_keys(tag): + values[lookup].append(value) + return archived_subtree.filter(Q(value__in=values["value"]) | Q(external_id__in=values["external_id"])).exists() + + def _report_archived_subtree_matches( + self, chunk: list[TagItem], first_row: int, archived_subtree: models.QuerySet[Tag] + ) -> None: + """ + Add one error for each row in the chunk that matches a tag in `archived_subtree`. + + A row matching on several fields is reported on the first of them only. + """ + for position, tag in enumerate(chunk, start=first_row): + for field, value, lookup in self._row_keys(tag): + match = archived_subtree.filter(**{lookup: value}).first() + if match: + self.errors.append( + ArchivedTagCollisionError(position, field, value, under_archived=not match.archived) + ) + break + + def _reject_rows_targeting_archived_subtree(self, tags: list[TagItem]) -> bool: + """ + Add an error for each row whose value, id, previous_id or parent_id matches a tag in an archived subtree. + + Return whether any row was rejected. Matching happens in the DB, so it follows the column collation + exactly as the unique constraints and the actions' own lookups do. It costs one query when nothing + in the taxonomy is archived, otherwise one more per chunk of rows, plus up to four lookups per row + in a chunk that matched. + """ + archived_subtree = self.taxonomy.tag_set.in_archived_subtree() + if not archived_subtree.exists(): + return False + errors_before = len(self.errors) + for start in range(0, len(tags), self.IN_CHUNK_SIZE): + chunk = tags[start:start + self.IN_CHUNK_SIZE] + if self._chunk_matches_archived_subtree(chunk, archived_subtree): + self._report_archived_subtree_matches(chunk, start + 1, archived_subtree) + return len(self.errors) > errors_before + def _build_staging_actions(self, tags: list[TagItem]) -> None: """ Stage any tag whose current external_id is another rename row's @@ -330,7 +397,8 @@ def _end_state( if key in in_file or key in deleted: continue # What remains are existing tags the file never mentions and that are not - # deleted, which only happens with replace=False. A renamed tag is not among + # deleted: all of them with replace=False, and archived subtrees with + # replace=True. A renamed tag is not among # them: the loop above added it under its final id. These tags are kept # because they still count toward the depth and toward cycles. Each stays # under the same parent tag, recorded by that parent's final id. @@ -461,7 +529,12 @@ def generate_actions( Validates each action and create respective errors If `replace` is True, then creates the delete action for tags - that are in the existing taxonomy but not the new tags list. + that are in the existing taxonomy but not the new tags list, + leaving out archived subtrees. + + A row that targets a tag in an archived subtree is an error, + and no actions are built at all: they would only be planned against + tags the import may not touch. TODO: Join/reduce actions. Ex. A tag may have no changes, but then its parent needs to be updated because its parent is deleted. @@ -482,11 +555,18 @@ def generate_actions( # _validate_no_plain_row_targets_a_vacated_id). self._validate_no_plain_row_targets_a_vacated_id(tags) + if self._reject_rows_targeting_archived_subtree(tags): + return + tags_for_delete = {} if replace: + # An archived subtree is also left out of the export, so its absence from the file + # is no reason to delete it. + archived_subtree = self.taxonomy.tag_set.in_archived_subtree() tags_for_delete = { - self._get_tag_id(tag): tag for tag in self.taxonomy.tag_set.all() + self._get_tag_id(tag): tag + for tag in self.taxonomy.tag_set.exclude(pk__in=archived_subtree.values("pk")) } # external_id is case-insensitive in the DB, so both pops below # must match regardless of case too. diff --git a/src/openedx_tagging/import_export/parsers.py b/src/openedx_tagging/import_export/parsers.py index 656062c90..471d73cf0 100644 --- a/src/openedx_tagging/import_export/parsers.py +++ b/src/openedx_tagging/import_export/parsers.py @@ -12,6 +12,7 @@ from django.utils.translation import gettext as _ +from ..data import TagData from ..models import Taxonomy from .exceptions import ( EmptyCSVField, @@ -209,17 +210,24 @@ def _load_tags_for_export(cls, taxonomy: Taxonomy) -> list[dict]: The tags are ordered by hierarchy, first, parents and then children. `get_filtered_tags` is in charge of returning this in a hierarchical way. + + A tag whose parent is not among the loaded tags (e.g. an archived parent) is skipped, + along with its descendants, so that every exported `parent_id` refers to an exported tag. """ tags = taxonomy.get_filtered_tags().all() result = [] + exported: dict[str, TagData] = {} # Parents always precede their children in `tags`. for tag in tags: result_tag = { "id": tag["external_id"] or tag["_id"], "value": tag["value"], } if tag["parent_value"]: - parent_tag = next(t for t in tags if t["value"] == tag["parent_value"]) + parent_tag = exported.get(tag["parent_value"]) + if parent_tag is None: + continue result_tag["parent_id"] = parent_tag["external_id"] or parent_tag["_id"] + exported[tag["value"]] = tag result.append(result_tag) return result diff --git a/src/openedx_tagging/migrations/0023_archived_and_deletion_locked.py b/src/openedx_tagging/migrations/0023_archived_and_deletion_locked.py new file mode 100644 index 000000000..a17445050 --- /dev/null +++ b/src/openedx_tagging/migrations/0023_archived_and_deletion_locked.py @@ -0,0 +1,33 @@ +# Generated by Django 5.2.17 on 2026-10-07 23:12 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('oel_tagging', '0022_tag_external_id_not_null'), + ] + + operations = [ + migrations.AddField( + model_name='objecttag', + name='archived', + field=models.BooleanField(default=False, help_text='Hides this association from active use by default, without removing it.'), + ), + migrations.AddField( + model_name='objecttag', + name='deletion_locked', + field=models.BooleanField(default=False, help_text='Set when something outside the tagging app depends on this association; a delete archives it instead of removing it.'), + ), + migrations.AddField( + model_name='tag', + name='archived', + field=models.BooleanField(default=False, help_text='Hides this tag from active use by default, without removing it.'), + ), + migrations.AddField( + model_name='taxonomy', + name='archived', + field=models.BooleanField(default=False, help_text="Hides this taxonomy from active use by default, without removing it. Unlike disabling it with 'enabled', archiving marks the taxonomy as retired."), + ), + ] diff --git a/src/openedx_tagging/models/base.py b/src/openedx_tagging/models/base.py index f725ec778..c009acc3c 100644 --- a/src/openedx_tagging/models/base.py +++ b/src/openedx_tagging/models/base.py @@ -5,11 +5,11 @@ from __future__ import annotations import re -from typing import List, Self, cast +from typing import ClassVar, List, Self, cast -from django.core.exceptions import ValidationError +from django.core.exceptions import ObjectDoesNotExist, ValidationError from django.db import models -from django.db.models import F, Value +from django.db.models import F, Q, Value from django.db.models.functions import Concat, Length, Replace, Substr from django.utils.functional import cached_property from django.utils.translation import gettext_lazy as _ @@ -27,6 +27,78 @@ Lineage = List[str] +class TaxonomyQuerySet(models.QuerySet["Taxonomy"]): + """ + QuerySet for :class:`Taxonomy` with an opt-in filter for archived rows. + + The default ``Taxonomy.objects`` manager returns every row, including + archived ones. Plain ``Taxonomy.objects`` is the supported way to reach + archived taxonomies; code that should hide them calls :meth:`unarchived`. + """ + + def unarchived(self) -> Self: + """ + Return only taxonomies whose ``archived`` flag is False. + """ + return self.filter(archived=False) + + +class TagQuerySet(models.QuerySet["Tag"]): + """ + QuerySet for :class:`Tag` with an opt-in filter for archived rows. + + The default ``Tag.objects`` manager returns every row, including archived + ones. Plain ``Tag.objects`` is the supported way to reach archived tags; + code that should hide them calls :meth:`unarchived`. + """ + + def unarchived(self) -> Self: + """ + Return only tags that are not archived and whose taxonomy is not archived. + + Tags with no taxonomy are kept, so the taxonomy condition is an exclude. + """ + return self.filter(archived=False).exclude(taxonomy__archived=True) + + def in_archived_subtree(self) -> Self: + """ + Return only tags that are archived or that sit under an archived tag. + + Such a tag still holds its value and external_id, but is no longer part of its taxonomy. + """ + # Depth is capped, so a fixed number of ancestor joins covers every tag, and the + # query stays the same size however many tags are archived. + condition = Q(archived=True) + for depth in range(1, TAXONOMY_MAX_DEPTH + 1): + condition |= Q(**{"__".join(["parent"] * depth) + "__archived": True}) + return self.filter(condition) + + +class ObjectTagQuerySet(models.QuerySet["ObjectTag"]): + """ + QuerySet for :class:`ObjectTag` with an opt-in filter for archived rows. + + The default ``ObjectTag.objects`` manager returns every row, including + archived ones. Plain ``ObjectTag.objects`` is the supported way to reach + archived object tags; code that should hide them calls :meth:`unarchived`. + """ + + def unarchived(self) -> Self: + """ + Return only object tags that are not archived and whose tag, tag's + taxonomy, and own taxonomy are not archived. + + The tag and taxonomy are nullable (free text and deleted records), so + those conditions are excludes: rows with no tag or taxonomy are kept. + """ + return ( + self.filter(archived=False) + .exclude(tag__archived=True) + .exclude(tag__taxonomy__archived=True) + .exclude(taxonomy__archived=True) + ) + + class Tag(models.Model): """ Represents a single value in a list or tree of values which can be applied to a particular Open edX object. @@ -84,6 +156,15 @@ class Tag(models.Model): " Set automatically by save(); do not set manually." ), ) + archived = models.BooleanField( + default=False, + help_text=_("Hides this tag from active use by default, without removing it."), + ) + + objects = TagQuerySet.as_manager() + + # Set right after the class body, because it has to subclass Tag.DoesNotExist, which Django creates with the class. + IsArchived: ClassVar[type[ObjectDoesNotExist]] class Meta: indexes = [ @@ -147,20 +228,20 @@ def get_lineage(self) -> Lineage: @cached_property def child_count(self) -> int: """ - How many child tags this tag has in the taxonomy. + How many unarchived child tags this tag has in the taxonomy. """ if self.taxonomy and not self.taxonomy.allow_free_text: - return self.taxonomy.tag_set.filter(parent=self).count() + return self.taxonomy.tag_set.unarchived().filter(parent=self).count() return 0 @cached_property def descendant_count(self) -> int: """ - How many descendant tags this tag has in the taxonomy. + How many unarchived descendant tags this tag has in the taxonomy. """ if self.taxonomy and not self.taxonomy.allow_free_text: # depth__gt correctly excludes self; lineage prefix matches all descendants. - return self.taxonomy.tag_set.filter( + return self.taxonomy.tag_set.unarchived().filter( depth__gt=self.depth, lineage__startswith=self.lineage, ).count() @@ -210,8 +291,14 @@ def save(self, *args, **kwargs) -> None: super().save(*args, **kwargs) - # Cascade lineage (and depth, if it changed) to all descendants in a single UPDATE. - if old_values is not None and old_values["lineage"] and old_values["lineage"] != self.lineage: + # Cascade lineage (and depth, if it changed) to all descendants in a single UPDATE. Tags without a taxonomy + # have no tree to cascade into, and filtering on taxonomy=None would match unrelated taxonomy-less tags. + if ( + old_values is not None + and self.taxonomy_id is not None + and old_values["lineage"] + and old_values["lineage"] != self.lineage + ): depth_delta = self.depth - old_values["depth"] update_kwargs: dict = { # Expression to compute the new lineage for each descendent: @@ -225,8 +312,9 @@ def save(self, *args, **kwargs) -> None: } if depth_delta != 0: update_kwargs["depth"] = F("depth") + depth_delta - self.taxonomy.tag_set.filter( # type: ignore[union-attr] - lineage__startswith=old_values["lineage"] + Tag.objects.filter( + taxonomy=self.taxonomy, + lineage__startswith=old_values["lineage"], ).exclude(pk=self.pk).update( **update_kwargs ) @@ -248,6 +336,18 @@ def clean(self): raise ValidationError("Tag external ID cannot contain a TAB character.") +class TagIsArchived(Tag.DoesNotExist): + """ + Raised when a tag exists but is archived, or belongs to an archived taxonomy. + + Subclasses ``Tag.DoesNotExist`` so that code that treats an archived tag as missing keeps working, + while callers that care about the difference can catch this first. Reachable as ``Tag.IsArchived``. + """ + + +Tag.IsArchived = TagIsArchived + + class Taxonomy(models.Model): """ Represents a namespace and rules for a group of tags. @@ -311,6 +411,15 @@ class Taxonomy(models.Model): " taxonomy admins will not be permitted to add, edit, or delete its tags." ), ) + archived = models.BooleanField( + default=False, + help_text=_( + "Hides this taxonomy from active use by default, without removing it." + " Unlike disabling it with 'enabled', archiving marks the taxonomy as retired." + ), + ) + + objects = TaxonomyQuerySet.as_manager() class Meta: verbose_name_plural = "Taxonomies" @@ -358,6 +467,10 @@ def get_filtered_tags( Use `excluded_values` to exclude tags with that value (and their parents, if applicable) from the results. + Archived tags, and every tag of an archived taxonomy, are excluded. For free text taxonomies, values from + archived object tags are excluded. Raises Tag.DoesNotExist if `parent_tag_value` is missing, or its subclass + Tag.IsArchived if that tag or this taxonomy is archived. + Note: This is mostly an 'internal' API and generally code outside of openedx_tagging should use the APIs in openedx_tagging.api which in turn use this. """ @@ -387,6 +500,19 @@ def get_filtered_tags( else: raise ValueError("Unsupported depth value for get_filtered_tags()") + def _unarchived_tag_for_value(self, value: str) -> Tag: + """ + Like tag_for_value(), but also raises Tag.IsArchived if the tag or this taxonomy is archived. + + Tag.IsArchived subclasses Tag.DoesNotExist, so callers that treat archived tags as missing can ignore it. + + Kept separate from tag_for_value() because tagging an object needs to resolve archived tags. + """ + tag = self.tag_for_value(value) + if tag.archived or self.archived: + raise Tag.IsArchived(f"Tag '{value}' is archived or belongs to an archived taxonomy.") + return tag + def _get_filtered_tags_free_text( self, search_term: str | None, @@ -395,7 +521,7 @@ def _get_filtered_tags_free_text( Implementation of get_filtered_tags() for free text taxonomies. """ assert self.allow_free_text - qs: models.QuerySet = self.objecttag_set.all() + qs: models.QuerySet = self.objecttag_set.unarchived() if search_term: qs = qs.filter(_value__icontains=search_term) # Rename "_value" to "value" @@ -424,14 +550,15 @@ def _get_filtered_tags_one_level( """ # A closed, and possibly hierarchical taxonomy. We're just fetching a single "level" of tags. if parent_tag_value: - parent_tag = self.tag_for_value(parent_tag_value) - qs: models.QuerySet = self.tag_set.filter(parent_id=parent_tag.pk) + parent_tag = self._unarchived_tag_for_value(parent_tag_value) + qs: models.QuerySet = self.tag_set.unarchived().filter(parent_id=parent_tag.pk) # Use parent_tag.value not parent_tag_value because they may differ in case qs = qs.annotate(parent_value=Value(parent_tag.value)) else: - qs = self.tag_set.filter(parent=None) + qs = self.tag_set.unarchived().filter(parent=None) qs = qs.annotate(parent_value=Value(None, output_field=models.CharField())) - qs = qs.annotate(child_count=models.Count("children", distinct=True)) # type: ignore[no-redef] + child_count = models.Count("children", filter=models.Q(children__archived=False), distinct=True) + qs = qs.annotate(child_count=child_count) # type: ignore[no-redef] # Filter by search term: if search_term: qs = qs.filter(value__icontains=search_term) @@ -453,13 +580,13 @@ def _get_filtered_tags_deep( """ if parent_tag_value: # Get a subtree below this tag: - main_parent_tag = self.tag_for_value(parent_tag_value) - initial_qs = self.tag_set.filter( + main_parent_tag = self._unarchived_tag_for_value(parent_tag_value) + initial_qs = self.tag_set.unarchived().filter( lineage__startswith=main_parent_tag.lineage, depth__gt=main_parent_tag.depth, ) else: - initial_qs = self.tag_set.all() + initial_qs = self.tag_set.unarchived() if search_term: # We need to do an additional query to find all the tags that match the search term, then limit the @@ -524,6 +651,9 @@ def add_tag( Add new Tag to Taxonomy. If an existing Tag with the `tag_value` already exists in the Taxonomy, an exception is raised, otherwise the newly created Tag is returned + + Raises ValueError if the taxonomy is archived, or if `parent_tag_value` is an archived tag or under one. + A value or `external_id` held by an archived tag is refused with its own message. """ if self.allow_free_text: raise ValueError( @@ -535,10 +665,29 @@ def add_tag( "add_tag() doesn't work for read-only taxonomies. They cannot be modified." ) - if self.tag_set.filter(value__iexact=tag_value).exists(): + if self.archived: + raise ValueError( + "add_tag() doesn't work for archived taxonomies. They cannot be modified." + ) + + # Tag.clean strips both before saving, so compare what would be saved. + tag_value = tag_value.strip() + external_id = external_id.strip() if external_id else external_id + existing = self.tag_set.filter(value__iexact=tag_value).first() + if existing and existing.archived: + raise ValueError( + f"Tag value '{tag_value}' belongs to an archived tag in this taxonomy. Use a different value." + ) + if existing: raise ValueError(f"Tag with value '{tag_value}' already exists for taxonomy.") - if external_id and self.validate_external_id(external_id): + existing = self.tag_set.filter(external_id__iexact=external_id).first() if external_id else None + if existing and existing.archived: + raise ValueError( + f"Tag external_id '{external_id}' belongs to an archived tag in this taxonomy. " + "Use a different external_id." + ) + if existing: raise ValueError(f"Tag with external_id '{external_id}' already exists for taxonomy.") parent = None @@ -546,6 +695,10 @@ def add_tag( # Get parent tag from taxonomy, raises Tag.DoesNotExist if doesn't # belong to taxonomy parent = self.tag_set.get(value__iexact=parent_tag_value) + # The lineage holds the values of the parent and its ancestors, so this also finds an archived ancestor. + archived = self.tag_set.filter(archived=True, value__in=parent.lineage.split("\t")[:-1]).first() + if archived: + raise ValueError(f"Cannot add a tag under the archived tag '{archived.value}'.") tag = Tag.objects.create( taxonomy=self, value=tag_value, parent=parent, external_id=external_id @@ -558,6 +711,8 @@ def update_tag(self, tag: str, new_value: str) -> Tag: """ Update an existing Tag in Taxonomy and return it. Currently only supports updating the Tag's value. + + Raises ValueError if the taxonomy or the tag is archived. """ if self.allow_free_text: raise ValueError( @@ -569,9 +724,15 @@ def update_tag(self, tag: str, new_value: str) -> Tag: "update_tag() doesn't work for read-only taxonomies. They cannot be modified." ) + # Checked before the lookup so that every tag name is rejected the same way in an archived taxonomy. + if self.archived: + raise ValueError("update_tag() doesn't work for archived taxonomies or tags. They cannot be modified.") + # Update Tag instance with new value, raises Tag.DoesNotExist if # tag doesn't belong to taxonomy tag_to_update = self.tag_set.get(value__iexact=tag) + if tag_to_update.archived: + raise ValueError("update_tag() doesn't work for archived taxonomies or tags. They cannot be modified.") tag_to_update.value = new_value tag_to_update.save() return tag_to_update @@ -724,6 +885,19 @@ class ObjectTag(models.Model): " If the tag field is set, then tag.value takes precedence over this field." ), ) + archived = models.BooleanField( + default=False, + help_text=_("Hides this association from active use by default, without removing it."), + ) + deletion_locked = models.BooleanField( + default=False, + help_text=_( + "Set when something outside the tagging app depends on this association;" + " a delete archives it instead of removing it." + ), + ) + + objects = ObjectTagQuerySet.as_manager() class Meta: indexes = [ @@ -896,6 +1070,8 @@ def copy(self, object_tag: ObjectTag) -> Self: self.taxonomy = object_tag.taxonomy self.object_id = object_tag.object_id self.is_copied = object_tag.is_copied + self.archived = object_tag.archived + self.deletion_locked = object_tag.deletion_locked self._value = object_tag._value # pylint: disable=protected-access self._export_id = object_tag._export_id # pylint: disable=protected-access return self diff --git a/src/openedx_tagging/rest_api/v1/serializers.py b/src/openedx_tagging/rest_api/v1/serializers.py index 18dd34e59..6eb715de7 100644 --- a/src/openedx_tagging/rest_api/v1/serializers.py +++ b/src/openedx_tagging/rest_api/v1/serializers.py @@ -104,11 +104,11 @@ def get_tags_count(self, instance): """ Return the "tags_count" annotation if present. - Or just count the taxonomy's tags. + Or just count the taxonomy's unarchived tags. """ if hasattr(instance, 'tags_count'): return instance.tags_count - return instance.tag_set.count() + return instance.tag_set.unarchived().count() def get_can_tag_object(self, instance) -> bool | None: """ @@ -238,6 +238,7 @@ def validate_tag_value(value, context, original_value=None): """ Validates the incoming request early: - This tag is unique, not a duplicate. (The model only validates this if you call `full_clean()`.) + A value held by an archived tag gets its own message, since that tag is not shown in the taxonomy's tag lists. - There are no forbidden / reserved characters present. There is an additional model-side validation for this as well, but we are keeping this so we can validate the incoming request immediately. @@ -252,8 +253,13 @@ def validate_tag_value(value, context, original_value=None): if tag_id: queryset = queryset.exclude(pk=tag_id) - # Check if tag value already exists within this taxonomy. If so, raise a validation error. - if queryset.exists(): + existing = queryset.first() + if existing and existing.archived: + raise serializers.ValidationError( + f'Tag value "{value}" belongs to an archived tag in this taxonomy. Use a different value.', + code='unique', + ) + if existing: raise serializers.ValidationError( f'Tag value "{value}" already exists in this taxonomy.', code='unique' ) diff --git a/src/openedx_tagging/rest_api/v1/views.py b/src/openedx_tagging/rest_api/v1/views.py index 9bf57b1bb..48a37563b 100644 --- a/src/openedx_tagging/rest_api/v1/views.py +++ b/src/openedx_tagging/rest_api/v1/views.py @@ -55,13 +55,24 @@ TaxonomyTagUpdateBodySerializer, ) +ARCHIVED_TAXONOMY_MESSAGE = "This taxonomy is archived and cannot be edited." + + +def _get_taxonomy_including_archived(taxonomy_id: int) -> Taxonomy | None: + """ + Returns the Taxonomy with the given ID even if it is archived, or None if not found. + + This is the unfiltered lookup for write and delete actions. Read actions use the archive-aware `get_taxonomy`. + """ + return Taxonomy.objects.filter(pk=taxonomy_id).first() + @view_auth_classes class TaxonomyView(TaggingExceptionHandlerMixin, ModelViewSet): """ View to list, create, retrieve, update, delete, export or import Taxonomies. - TODO: We need to add a perform_update and call the api update function when is created. + TODO: `perform_update` needs to call the api update function when it is created. This is because it is necessary to call the model validations. (`full_clean()`). Currently those validations are not run, which means we could update `export_id` with any value through this api. @@ -228,19 +239,39 @@ class TaxonomyView(TaggingExceptionHandlerMixin, ModelViewSet): permission_classes = [TaxonomyObjectPermissions] pagination_class = TaxonomyPagination + # Actions for which `get_object` also resolves archived taxonomies. Writes reject them with a 400, and + # destroy keeps its existing behavior. + ACTIONS_THAT_RESOLVE_ARCHIVED = ("update", "partial_update", "destroy", "update_import", "plan_update_import") + def get_object(self) -> Taxonomy: """ Return the requested taxonomy object, if the user has appropriate permissions. + + Archived taxonomies are not found, except for the write and delete actions, which resolve them so that + the write can be rejected with an explanation and destroy keeps its behavior for archived taxonomies. """ pk = int(self.kwargs["pk"]) - taxonomy = get_taxonomy(pk) + if self.action in self.ACTIONS_THAT_RESOLVE_ARCHIVED: + taxonomy = _get_taxonomy_including_archived(pk) + else: + taxonomy = get_taxonomy(pk) if not taxonomy: raise Http404("Taxonomy not found") self.check_object_permissions(self.request, taxonomy) return taxonomy + def perform_update(self, serializer) -> None: + """ + Update the taxonomy, unless it is archived. + + `get_object` resolves archived taxonomies for this action, so the check lives here. + """ + if serializer.instance.archived: + raise ValidationError(ARCHIVED_TAXONOMY_MESSAGE) + super().perform_update(serializer) + def get_queryset(self) -> models.QuerySet: """ Return a list of taxonomies. @@ -256,7 +287,7 @@ def get_queryset(self) -> models.QuerySet: enabled = query_params.data.get("enabled", None) qs = get_taxonomies(enabled) - qs = qs.annotate(tags_count=models.Count("tag", distinct=True)) + qs = qs.annotate(tags_count=models.Count("tag", filter=models.Q(tag__archived=False), distinct=True)) return qs def perform_create(self, serializer) -> None: @@ -340,7 +371,7 @@ def create_import(self, request: Request, **_kwargs) -> Response: def update_import(self, request: Request, **_kwargs) -> Response: """ Imports tags from the uploaded file to an already created taxonomy, - overwriting any existing tags. + overwriting any existing tags. Archived taxonomies cannot be imported into. """ body = TaxonomyImportBodySerializer(data=request.data) body.is_valid(raise_exception=True) @@ -349,6 +380,8 @@ def update_import(self, request: Request, **_kwargs) -> Response: parser_format = body.validated_data["parser_format"] taxonomy = self.get_object() + if taxonomy.archived: + raise ValidationError(ARCHIVED_TAXONOMY_MESSAGE) try: import_success, task, _plan = import_tags(taxonomy, file, parser_format, replace=True) @@ -366,7 +399,7 @@ def update_import(self, request: Request, **_kwargs) -> Response: def plan_update_import(self, request: Request, **_kwargs) -> Response: """ Plan import tags from the uploaded file to an already created taxonomy, - overwriting any existing tags. + overwriting any existing tags. Archived taxonomies cannot be imported into. """ body = TaxonomyImportBodySerializer(data=request.data) body.is_valid(raise_exception=True) @@ -375,6 +408,8 @@ def plan_update_import(self, request: Request, **_kwargs) -> Response: parser_format = body.validated_data["parser_format"] taxonomy = self.get_object() + if taxonomy.archived: + raise ValidationError(ARCHIVED_TAXONOMY_MESSAGE) try: import_success, task, plan = import_tags(taxonomy, file, parser_format, replace=True, plan_only=True) @@ -800,11 +835,17 @@ def get_taxonomy(self) -> Taxonomy: """ Get the taxonomy from `pk` or raise 404. + Archived taxonomies are not found, except for POST, PUT and PATCH, which resolve them so that + the tag write is rejected with an explanation instead of a 404. GET and DELETE do not. + The current taxonomy is cached in the view. """ if not self._taxonomy: taxonomy_id = self.kwargs.get("pk") - taxonomy = get_taxonomy(int(taxonomy_id)) if taxonomy_id else None + if taxonomy_id and self.request.method in ("POST", "PUT", "PATCH"): + taxonomy = _get_taxonomy_including_archived(int(taxonomy_id)) + else: + taxonomy = get_taxonomy(int(taxonomy_id)) if taxonomy_id else None if not taxonomy: raise Http404("Taxonomy not found") self.check_object_permissions(self.request, taxonomy) @@ -860,11 +901,14 @@ def get_queryset(self) -> TagDataQuerySet: else: depth = 1 # Otherwise (default), load a single level of results only. - results = taxonomy.get_filtered_tags( - parent_tag_value=parent_tag_value, - search_term=search_term, - depth=depth, - ) + try: + results = taxonomy.get_filtered_tags( + parent_tag_value=parent_tag_value, + search_term=search_term, + depth=depth, + ) + except TagDoesNotExist as e: + raise Http404("Parent tag not found") from e if depth == 1: # We're already returning just a single level. It will be paginated normally. if include_counts: diff --git a/tests/openedx_tagging/import_export/test_api.py b/tests/openedx_tagging/import_export/test_api.py index 9bf8e392b..c5fc13e94 100644 --- a/tests/openedx_tagging/import_export/test_api.py +++ b/tests/openedx_tagging/import_export/test_api.py @@ -1084,6 +1084,113 @@ def test_import_same_value_without_external_id(self) -> None: ) assert result + def _snapshot(self, taxonomy: Taxonomy) -> set[tuple]: + return set(taxonomy.tag_set.values_list("external_id", "value", "parent__external_id", "archived")) + + def _json_file(self, tags: list[dict]) -> BytesIO: + return BytesIO(json.dumps({"tags": tags}).encode()) + + def test_import_archived_value_changes_nothing(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, external_id="tag_3").update(archived=True) + before = self._snapshot(self.taxonomy) + file = self._json_file([ + {"id": "tag_50", "value": "Fifty"}, + {"id": "tag_51", "value": "Tag 3"}, + {"id": "tag_52", "value": "Fifty Two"}, + {"id": "tag_53", "value": "Fifty Three", "parent_id": "tag_50"}, + ]) + + result, _task, _plan = import_export_api.import_tags(self.taxonomy, file, self.parser_format) + + assert not result + assert "belongs to an archived tag" in import_export_api.get_last_import_log(self.taxonomy) + assert self._snapshot(self.taxonomy) == before + + def test_import_archived_id_does_not_revive_tag(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, external_id="tag_3").update(archived=True) + before = self._snapshot(self.taxonomy) + file = self._json_file([{"id": "tag_3", "value": "Revived Tag 3"}, {"id": "tag_50", "value": "Fifty"}]) + + result, _task, _plan = import_export_api.import_tags(self.taxonomy, file, self.parser_format) + + assert not result + assert "Row 1's id (tag_3) belongs to an archived tag" in import_export_api.get_last_import_log(self.taxonomy) + assert self._snapshot(self.taxonomy) == before + + def test_replace_import_promotes_archived_child_of_deleted_tag(self) -> None: + # The alternative is the delete cascading to the archived child. + Tag.objects.filter(taxonomy=self.taxonomy, external_id="tag_4").update(archived=True) + file = self._json_file([ + {"id": "tag_1", "value": "Tag 1"}, + {"id": "tag_2", "value": "Tag 2", "parent_id": "tag_1"}, + ]) + + result, _task, _plan = import_export_api.import_tags( + self.taxonomy, file, self.parser_format, replace=True, + ) + + assert result + assert not self.taxonomy.tag_set.filter(external_id="tag_3").exists() + archived = self.taxonomy.tag_set.get(external_id="tag_4") + assert archived.archived + assert archived.parent is None + + def _taxonomy_with_archived_tags(self, name: str) -> Taxonomy: + """ + Four root tags, two of them archived, and an unarchived child under one of the archived ones. + """ + taxonomy = Taxonomy.objects.create(name=name, export_id=name.lower().replace(" ", "_")) + for value in ("Kept One", "Kept Two", "Archived Root", "Archived Parent"): + taxonomy.add_tag(value) + taxonomy.add_tag("Child Of Archived", parent_tag_value="Archived Parent") + taxonomy.tag_set.filter(value__in=["Archived Root", "Archived Parent"]).update(archived=True) + return taxonomy + + @ddt.data(ParserFormat.JSON, ParserFormat.CSV) + def test_export_omits_archived_tag(self, parser_format: ParserFormat) -> None: + taxonomy = Taxonomy.objects.create(name="Three roots") + for value in ("Kept One", "Kept Two", "Archived Root"): + taxonomy.add_tag(value) + taxonomy.tag_set.filter(value="Archived Root").update(archived=True) + + output = import_export_api.export_tags(taxonomy, parser_format) + + assert "Kept One" in output + assert "Kept Two" in output + assert "Archived Root" not in output + + @ddt.data(ParserFormat.JSON, ParserFormat.CSV) + def test_export_then_import_into_empty_taxonomy(self, parser_format: ParserFormat) -> None: + source = self._taxonomy_with_archived_tags("Source") + output = import_export_api.export_tags(source, parser_format) + target = Taxonomy.objects.create(name="Target", export_id="target") + + result, _task, _plan = import_export_api.import_tags(target, BytesIO(output.encode()), parser_format) + + assert result + # The export leaves out a child of an archived tag too, so that every parent_id resolves. + assert set(target.tag_set.values_list("value", flat=True)) == {"Kept One", "Kept Two"} + + @ddt.data(ParserFormat.JSON, ParserFormat.CSV) + def test_replace_reimport_of_own_export_keeps_archived_tags(self, parser_format: ParserFormat) -> None: + taxonomy = self._taxonomy_with_archived_tags("Own export") + output = import_export_api.export_tags(taxonomy, parser_format) + + result, _task, _plan = import_export_api.import_tags( + taxonomy, BytesIO(output.encode()), parser_format, replace=True, + ) + + assert result + assert set(taxonomy.tag_set.filter(archived=True).values_list("value", flat=True)) == { + "Archived Root", "Archived Parent", + } + assert set(taxonomy.tag_set.unarchived().values_list("value", flat=True)) == { + "Kept One", "Kept Two", "Child Of Archived", + } + child = taxonomy.tag_set.get(value="Child Of Archived") + assert child.parent is not None + assert child.parent.value == "Archived Parent" + class TestImportAcceptanceScenarios(TestCase): """ diff --git a/tests/openedx_tagging/import_export/test_import_plan.py b/tests/openedx_tagging/import_export/test_import_plan.py index 206e8cee9..86e9211e5 100644 --- a/tests/openedx_tagging/import_export/test_import_plan.py +++ b/tests/openedx_tagging/import_export/test_import_plan.py @@ -1,12 +1,14 @@ """ Test for import_plan functions """ +from unittest.mock import patch + import ddt # type: ignore[import] from django.db import transaction from django.test.testcases import TestCase -from openedx_tagging.import_export.actions import CreateTag -from openedx_tagging.import_export.exceptions import TagImportError +from openedx_tagging.import_export.actions import CreateTag, DeleteTag +from openedx_tagging.import_export.exceptions import ArchivedTagCollisionError, TagImportError from openedx_tagging.import_export.import_plan import TagImportPlan, TagItem from openedx_tagging.models import Tag, TagImportTask, Taxonomy @@ -1053,3 +1055,118 @@ def test_generate_actions_rejects_duplicate_final_id_regardless_of_order(self) - error = str(self.import_plan.errors[0]) self.assertIn("tag_1", error) self.assertIn("file rows 1, 3", error) + + def _archive(self, external_id: str) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, external_id=external_id).update(archived=True) + + @ddt.data( + (TagItem(id='tag_50', value='Tag 3'), "value (Tag 3)", "Use a different value."), + (TagItem(id='tag_50', value=' Tag 3 '), "value (Tag 3)", "Use a different value."), + (TagItem(id='TAG_3', value='Brand New'), "id (TAG_3)", "Use a different id."), + (TagItem(id='tag_50', value='Brand New', parent_id='tag_3'), "parent_id (tag_3)", "Use a different parent_id."), + ( + TagItem(id='tag_50', value='Brand New', previous_id='tag_3'), + "previous_id (tag_3)", + "An import cannot change it.", + ), + ) + @ddt.unpack + def test_generate_actions_archived_tag_rejected(self, row: TagItem, field_and_value: str, advice: str) -> None: + self._archive('tag_3') + self.import_plan.generate_actions(tags=[row]) + assert len(self.import_plan.errors) == 1 + error = self.import_plan.errors[0] + assert isinstance(error, ArchivedTagCollisionError) + assert str(error) == f"Row 1's {field_and_value} belongs to an archived tag in this taxonomy. {advice}" + assert not self.import_plan.actions + + def test_generate_actions_one_archived_error_per_row(self) -> None: + self._archive('tag_3') + self.import_plan.generate_actions(tags=[TagItem(id='tag_3', value='Tag 3', parent_id='tag_3')]) + assert [str(error) for error in self.import_plan.errors] == [ + "Row 1's value (Tag 3) belongs to an archived tag in this taxonomy. Use a different value." + ] + + def test_generate_actions_reports_every_archived_collision(self) -> None: + self._archive('tag_3') + tags = [ + TagItem(id='tag_50', value='Fifty'), + TagItem(id='tag_51', value='Tag 3'), + TagItem(id='tag_52', value='Fifty Two'), + TagItem(id='tag_3', value='Fifty Three'), + ] + self.import_plan.generate_actions(tags=tags) + errors = [str(error) for error in self.import_plan.errors] + assert len(errors) == 2 + assert errors[0].startswith("Row 2's value (Tag 3)") + assert errors[1].startswith("Row 4's id (tag_3)") + assert not self.import_plan.actions + + @ddt.data( + ('Photosynthesis', 'PHOTOSYNTHESIS'), # every supported collation folds these + ('Straße', 'STRASSE'), # casefold folds these; SQLite NOCASE does not + ('Café', 'Cafe'), # MySQL's collation folds these; casefold does not + ) + @ddt.unpack + def test_generate_actions_archived_match_follows_db_collation(self, archived_value: str, row_value: str) -> None: + archived = self.taxonomy.add_tag(archived_value) + Tag.objects.filter(pk=archived.pk).update(archived=True) + collides = self.taxonomy.tag_set.filter(value=row_value).exists() + + self.import_plan.generate_actions(tags=[TagItem(id='tag_50', value=row_value)]) + + # On SQLite the Café case collides on neither side; it only bites on MySQL. + archived_errors = [e for e in self.import_plan.errors if isinstance(e, ArchivedTagCollisionError)] + assert bool(archived_errors) == collides + + @ddt.data( + ('id', TagItem(id='tag_4', value='Renamed Tag 4')), + ('value', TagItem(id='tag_50', value='Tag 4')), + ('parent_id', TagItem(id='tag_50', value='Fifty', parent_id='tag_4')), + ) + @ddt.unpack + def test_generate_actions_tag_under_archived_tag_rejected(self, field: str, row: TagItem) -> None: + self._archive('tag_3') # tag_4 stays unarchived, under it + self.import_plan.generate_actions(tags=[row]) + value = getattr(row, field) + assert [str(error) for error in self.import_plan.errors] == [ + f"Row 1's {field} ({value}) belongs to a tag under an archived tag in this taxonomy. " + f"Use a different {field}." + ] + assert not self.import_plan.actions + + def test_generate_actions_replace_keeps_archived_subtree(self) -> None: + self._archive('tag_3') + tags = [TagItem(id='tag_1', value='Tag 1'), TagItem(id='tag_2', value='Tag 2', parent_id='tag_1')] + self.import_plan.generate_actions(tags=tags, replace=True) + assert not self.import_plan.errors + deleted_ids = {action.tag.id for action in self.import_plan.actions if isinstance(action, DeleteTag)} + # Neither the archived tag_3 nor tag_4, which is unarchived but sits under it, is deleted. + assert not deleted_ids + + @ddt.data(False, True) + def test_generate_actions_with_many_archived_tags(self, replace: bool) -> None: + # More archived tags than SQLite allows terms in one expression tree. + Tag.objects.bulk_create( + Tag(taxonomy=self.taxonomy, value=f"Old {i}", external_id=f"old_{i}", lineage=f"Old {i}\t", archived=True) + for i in range(1100) + ) + tags = [TagItem(id=tag.external_id, value=tag.value) for tag in self.taxonomy.tag_set.unarchived()] + tags.append(TagItem(id='tag_50', value='Fifty')) + + self.import_plan.generate_actions(tags=tags, replace=replace) + + assert not self.import_plan.errors + assert not [action for action in self.import_plan.actions if isinstance(action, DeleteTag)] + + def test_generate_actions_archived_collision_in_last_chunk(self) -> None: + self._archive('tag_3') + tags = [TagItem(id=f'tag_{i}', value=f'New {i}') for i in range(50, 55)] + tags.append(TagItem(id='tag_60', value='Tag 3')) + + with patch.object(TagImportPlan, 'IN_CHUNK_SIZE', 2): + self.import_plan.generate_actions(tags=tags) + + assert [str(error) for error in self.import_plan.errors] == [ + "Row 6's value (Tag 3) belongs to an archived tag in this taxonomy. Use a different value." + ] diff --git a/tests/openedx_tagging/import_export/test_parsers.py b/tests/openedx_tagging/import_export/test_parsers.py index a7c55f106..2c028e78f 100644 --- a/tests/openedx_tagging/import_export/test_parsers.py +++ b/tests/openedx_tagging/import_export/test_parsers.py @@ -11,7 +11,7 @@ from openedx_tagging.import_export.exceptions import InvalidJSONField, TagParserError from openedx_tagging.import_export.parsers import CSVParser, JSONParser, Parser, ParserFormat, get_parser -from openedx_tagging.models import Taxonomy +from openedx_tagging.models import Tag, Taxonomy from .mixins import TestImportExportMixin @@ -218,6 +218,20 @@ def test_export_data(self) -> None: if tag.get("parent_id"): assert tag.get("parent_id") == taxonomy_tag.parent.external_id + def test_export_data_omits_archived_tag(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, external_id="tag_4").update(archived=True) + tags = json.loads(JSONParser.export(self.taxonomy)).get("tags") + assert [tag["id"] for tag in tags] == ["tag_1", "tag_2", "tag_3"] + + def test_export_data_omits_children_of_archived_tag(self) -> None: + tag_2 = Tag.objects.get(taxonomy=self.taxonomy, external_id="tag_2") + Tag.objects.create(taxonomy=self.taxonomy, value="Tag 5", external_id="tag_5", parent=tag_2) + Tag.objects.filter(taxonomy=self.taxonomy, external_id="tag_1").update(archived=True) + # The unarchived descendants of the archived tag_1 are omitted too, so no parent_id dangles. + tags = json.loads(JSONParser.export(self.taxonomy)).get("tags") + assert [tag["id"] for tag in tags] == ["tag_3", "tag_4"] + assert tags[1]["parent_id"] == "tag_3" + def test_import_with_export_output(self) -> None: output = JSONParser.export(self.taxonomy) json_file = BytesIO(output.encode()) @@ -398,6 +412,21 @@ def test_parse_tags(self) -> None: index + CSVParser.inital_row ) + def test_export_data_omits_archived_tag(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, external_id="tag_4").update(archived=True) + tags, errors = CSVParser.parse_import(BytesIO(CSVParser.export(self.taxonomy).encode())) + assert not errors + assert [tag.id for tag in tags] == ["tag_1", "tag_2", "tag_3"] + + def test_export_data_omits_children_of_archived_tag(self) -> None: + tag_2 = Tag.objects.get(taxonomy=self.taxonomy, external_id="tag_2") + Tag.objects.create(taxonomy=self.taxonomy, value="Tag 5", external_id="tag_5", parent=tag_2) + Tag.objects.filter(taxonomy=self.taxonomy, external_id="tag_1").update(archived=True) + # The unarchived descendants of the archived tag_1 are omitted too, so no parent_id dangles. + tags, errors = CSVParser.parse_import(BytesIO(CSVParser.export(self.taxonomy).encode())) + assert not errors + assert [(tag.id, tag.parent_id) for tag in tags] == [("tag_3", None), ("tag_4", "tag_3")] + def test_import_with_export_output(self) -> None: output = CSVParser.export(self.taxonomy) csv_file = BytesIO(output.encode()) diff --git a/tests/openedx_tagging/test_api.py b/tests/openedx_tagging/test_api.py index 7c9931925..71b56d396 100644 --- a/tests/openedx_tagging/test_api.py +++ b/tests/openedx_tagging/test_api.py @@ -139,6 +139,135 @@ def test_get_tags(self) -> None: " Protista (children: 0)", ] + def test_get_tags_excludes_archived(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value__in=["Fungi", "Cnidaria"]).update(archived=True) + values = [row["value"] for row in tagging_api.get_tags(self.taxonomy)] + assert "Fungi" not in values + assert "Cnidaria" not in values + assert "Plantae" in values + # The tags are still in the database, and parents no longer count them as children: + assert self.taxonomy.tag_set.filter(value="Fungi").exists() + assert "Eukaryota (children: 4)" in pretty_format_tags(tagging_api.get_tags(self.taxonomy), parent=False) + + def test_get_root_tags_excludes_archived(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value="Archaea").update(archived=True) + assert pretty_format_tags(tagging_api.get_root_tags(self.taxonomy), parent=False) == [ + 'Bacteria (children: 2)', + 'Eukaryota (children: 5)', + ] + + def test_get_children_tags_excludes_archived(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value="Cnidaria").update(archived=True) + values = [row["value"] for row in tagging_api.get_children_tags(self.taxonomy, "Animalia")] + assert values == ["Arthropoda", "Chordata", "Ctenophora", "Gastrotrich", "Placozoa", "Porifera"] + # The parent's own child count doesn't include the archived child either: + children = pretty_format_tags(tagging_api.get_children_tags(self.taxonomy, "Eukaryota"), parent=False) + assert " Animalia (children: 6)" in children + + def test_get_children_tags_archived_parent(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value="Chordata").update(archived=True) + with self.assertRaises(Tag.IsArchived): + tagging_api.get_children_tags(self.taxonomy, "Chordata") + with self.assertRaises(Tag.IsArchived): # The full-depth query behaves the same way + self.taxonomy.get_filtered_tags(parent_tag_value="Chordata") + + def test_archived_parent_error_is_a_does_not_exist_error(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value="Chordata").update(archived=True) + # Existing handlers for a missing tag keep working for an archived one: + with self.assertRaises(Tag.DoesNotExist): + tagging_api.get_children_tags(self.taxonomy, "Chordata") + # But a tag that never existed is not reported as archived: + with self.assertRaises(Tag.DoesNotExist) as missing: + tagging_api.get_children_tags(self.taxonomy, "No such tag") + assert not isinstance(missing.exception, Tag.IsArchived) + + def test_search_tags_excludes_archived(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value="Eubacteria").update(archived=True) + result = tagging_api.search_tags(self.taxonomy, search_term='eU') + assert pretty_format_tags(result, parent=False) == [ + 'Archaea (children: 1)', + ' Euryarchaeida (children: 0)', + 'Eukaryota (children: 0)', + ] + + def test_search_tags_exclude_object_ignores_archived_object_tag(self) -> None: + tagging_api.tag_object(object_id="obj1", taxonomy=self.taxonomy, tags=["Fungi"]) + values = [row["value"] for row in tagging_api.search_tags(self.taxonomy, "Fungi", exclude_object_id="obj1")] + assert values == [] # Applied to the object, so it is excluded from its autocomplete + ObjectTag.objects.filter(object_id="obj1").update(archived=True) + values = [row["value"] for row in tagging_api.search_tags(self.taxonomy, "Fungi", exclude_object_id="obj1")] + assert "Fungi" in values # The ancestor "Eukaryota" is included too + + def test_archived_taxonomy_hides_its_tags(self) -> None: + Taxonomy.objects.filter(pk=self.taxonomy.pk).update(archived=True) + self.taxonomy.refresh_from_db() + assert not list(tagging_api.get_tags(self.taxonomy)) + assert not list(tagging_api.get_root_tags(self.taxonomy)) + assert not list(tagging_api.search_tags(self.taxonomy, search_term='eU')) + with self.assertRaises(Tag.DoesNotExist): + tagging_api.get_children_tags(self.taxonomy, "Animalia") + + def test_get_taxonomy_excludes_archived(self) -> None: + Taxonomy.objects.filter(pk=self.taxonomy.pk).update(archived=True) + assert tagging_api.get_taxonomy(self.taxonomy.pk) is None + + def test_get_taxonomy_by_export_id_excludes_archived(self) -> None: + Taxonomy.objects.filter(pk=self.taxonomy.pk).update(archived=True) + assert tagging_api.get_taxonomy_by_export_id("life_on_earth") is None + + def test_get_taxonomies_excludes_archived(self) -> None: + Taxonomy.objects.filter(pk=self.taxonomy.pk).update(archived=True) + assert self.taxonomy not in tagging_api.get_taxonomies(enabled=None) + assert self.taxonomy not in tagging_api.get_taxonomies() + + def test_get_tags_free_text_excludes_archived(self) -> None: + tagging_api.tag_object(object_id="obj1", taxonomy=self.free_text_taxonomy, tags=["keep", "hide"]) + ObjectTag.objects.filter(taxonomy=self.free_text_taxonomy, _value="hide").update(archived=True) + assert [row["value"] for row in tagging_api.get_tags(self.free_text_taxonomy)] == ["keep"] + + @ddt.data("Archaea", "archaea", " Archaea ") + def test_add_tag_with_archived_value_is_refused(self, value: str) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value="Archaea").update(archived=True) + with pytest.raises(ValueError) as exc: + tagging_api.add_tag_to_taxonomy(self.taxonomy, value) + assert str(exc.value) == ( + f"Tag value '{value.strip()}' belongs to an archived tag in this taxonomy. Use a different value." + ) + assert self.taxonomy.tag_set.filter(value="Archaea").count() == 1 + + def test_add_tag_with_archived_external_id_is_refused(self) -> None: + archaea = self.taxonomy.tag_set.get(value="Archaea") + Tag.objects.filter(pk=archaea.pk).update(archived=True) + with pytest.raises(ValueError) as exc: + tagging_api.add_tag_to_taxonomy(self.taxonomy, "New Tag", external_id=archaea.external_id) + assert str(exc.value) == ( + f"Tag external_id '{archaea.external_id}' belongs to an archived tag in this taxonomy. " + "Use a different external_id." + ) + assert not self.taxonomy.tag_set.filter(value="New Tag").exists() + + def test_add_tag_with_unarchived_duplicate_external_id_keeps_message(self) -> None: + archaea = self.taxonomy.tag_set.get(value="Archaea") + with pytest.raises(ValueError) as exc: + tagging_api.add_tag_to_taxonomy(self.taxonomy, "New Tag", external_id=archaea.external_id) + assert str(exc.value) == f"Tag with external_id '{archaea.external_id}' already exists for taxonomy." + + def test_add_tag_under_a_tag_under_an_archived_tag_is_refused(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value="Eukaryota").update(archived=True) + with pytest.raises(ValueError) as exc: + tagging_api.add_tag_to_taxonomy(self.taxonomy, "New Tag", parent_tag_value="Animalia") + assert str(exc.value) == "Cannot add a tag under the archived tag 'Eukaryota'." + assert not self.taxonomy.tag_set.filter(value="New Tag").exists() + + def test_add_tag_with_unarchived_duplicate_keeps_message(self) -> None: + with pytest.raises(ValueError) as exc: + tagging_api.add_tag_to_taxonomy(self.taxonomy, "Archaea") + assert str(exc.value) == "Tag with value 'Archaea' already exists for taxonomy." + + def test_add_tag_with_free_value_succeeds(self) -> None: + tag = tagging_api.add_tag_to_taxonomy(self.taxonomy, "Respiration") + assert tag.value == "Respiration" + def test_get_tags_system(self) -> None: assert pretty_format_tags(tagging_api.get_tags(self.read_only_taxonomy), parent=False) == [ "System Tag 1 (children: 0)", @@ -819,6 +948,110 @@ def test_get_object_tag_counts_deleted_disabled(self) -> None: tagging_api.add_tag_to_taxonomy(self.taxonomy, "DPANN", parent_tag_value="Archaea") assert tagging_api.get_object_tag_counts("object_*") == {obj1: 2, obj2: 2} + def test_get_object_tag_counts_excludes_archived(self) -> None: + obj1, obj2 = "object_id1", "object_id2" + tagging_api.tag_object(object_id=obj1, taxonomy=self.taxonomy, tags=["DPANN"]) + tagging_api.tag_object(object_id=obj2, taxonomy=self.taxonomy, tags=["Chordata"]) + ObjectTag.objects.filter(object_id=obj1).update(archived=True) # obj1's only association + + with self.assertNumQueries(1): + assert tagging_api.get_object_tag_counts("object_*") == {obj2: 1} + with self.assertNumQueries(1): + assert tagging_api.get_object_tag_counts("object_*", count_implicit=True) == {obj2: 3} + + def test_add_usage_counts_excludes_archived(self) -> None: + self.taxonomy.allow_multiple = True + self.taxonomy.save() + tagging_api.tag_object(object_id="obj1", taxonomy=self.taxonomy, tags=["DPANN"]) + tagging_api.tag_object(object_id="obj2", taxonomy=self.taxonomy, tags=["DPANN"]) + ObjectTag.objects.filter(object_id="obj2").update(archived=True) + + tag_data = tagging_api.get_children_tags(self.taxonomy, "Archaea") + with self.assertNumQueries(2): # One for the tag data and one for the usage counts + result = tagging_api.add_usage_counts(self.taxonomy, tag_data) + assert {row["value"]: row["usage_count"] for row in result} == { + "DPANN": 1, + "Euryarchaeida": 0, + "Proteoarchaeota": 0, + } + + def test_add_usage_counts_archived_only_association_is_zero(self) -> None: + tagging_api.tag_object(object_id="obj1", taxonomy=self.taxonomy, tags=["DPANN"]) # DPANN's only association + ObjectTag.objects.filter(object_id="obj1").update(archived=True) + + tag_data = tagging_api.get_children_tags(self.taxonomy, "Archaea") + with self.assertNumQueries(2): # One for the tag data and one for the usage counts + result = tagging_api.add_usage_counts(self.taxonomy, tag_data) + assert {row["value"]: row["usage_count"] for row in result}["DPANN"] == 0 + + def test_get_object_tags_excludes_archived_object_tag(self) -> None: + tagging_api.tag_object(object_id="obj1", taxonomy=self.taxonomy, tags=["DPANN"]) + ObjectTag.objects.filter(object_id="obj1").update(archived=True) + assert not tagging_api.get_object_tags("obj1").exists() + + def test_get_object_tags_excludes_archived_tag(self) -> None: + tagging_api.tag_object(object_id="obj1", taxonomy=self.taxonomy, tags=["DPANN"]) + Tag.objects.filter(taxonomy=self.taxonomy, value="DPANN").update(archived=True) + assert not ObjectTag.objects.get(object_id="obj1").archived # Only the tag is archived + assert not tagging_api.get_object_tags("obj1").exists() + + def test_get_object_tags_excludes_archived_taxonomy(self) -> None: + tagging_api.tag_object(object_id="obj1", taxonomy=self.taxonomy, tags=["DPANN"]) + Taxonomy.objects.filter(pk=self.taxonomy.pk).update(archived=True) + assert not ObjectTag.objects.get(object_id="obj1").archived # Neither the association... + assert not get_tag("DPANN").archived # ...nor the tag is archived + assert not tagging_api.get_object_tags("obj1").exists() + + def test_get_object_tags_excludes_archived_free_text_taxonomy(self) -> None: + # Only the taxonomy is archived: a free text object tag has no tag, so no other condition can hide it. + self.free_text_taxonomy.enabled = True + self.free_text_taxonomy.save() + tagging_api.tag_object(object_id="obj1", taxonomy=self.free_text_taxonomy, tags=["free"]) + assert tagging_api.get_object_tags("obj1").exists() + assert tagging_api.get_object_tag_counts("obj1") == {"obj1": 1} + + Taxonomy.objects.filter(pk=self.free_text_taxonomy.pk).update(archived=True) + assert not ObjectTag.objects.get(object_id="obj1").archived + assert not tagging_api.get_object_tags("obj1").exists() + assert tagging_api.get_object_tag_counts("obj1") == {} + + def test_archived_records_still_resolve_through_foreign_keys(self) -> None: + tagging_api.tag_object(object_id="obj1", taxonomy=self.taxonomy, tags=["DPANN"]) + Tag.objects.filter(taxonomy=self.taxonomy, value="DPANN").update(archived=True) + Taxonomy.objects.filter(pk=self.taxonomy.pk).update(archived=True) + object_tag = ObjectTag.objects.get(object_id="obj1") + assert object_tag.tag is not None + assert object_tag.tag.archived + assert object_tag.taxonomy is not None + assert object_tag.taxonomy.archived + assert object_tag.value == "DPANN" + + def test_default_managers_include_archived(self) -> None: + tagging_api.tag_object(object_id="obj1", taxonomy=self.taxonomy, tags=["DPANN"]) + tag = get_tag("DPANN") + Tag.objects.filter(pk=tag.pk).update(archived=True) + Taxonomy.objects.filter(pk=self.read_only_taxonomy.pk).update(archived=True) + ObjectTag.objects.filter(object_id="obj1").update(archived=True) + assert Tag.objects.filter(pk=tag.pk).exists() + assert not Tag.objects.unarchived().filter(pk=tag.pk).exists() + assert Taxonomy.objects.filter(pk=self.read_only_taxonomy.pk).exists() + assert not Taxonomy.objects.unarchived().filter(pk=self.read_only_taxonomy.pk).exists() + assert ObjectTag.objects.filter(object_id="obj1").exists() + assert not ObjectTag.objects.unarchived().filter(object_id="obj1").exists() + + def test_update_tag_in_taxonomy_archived_tag(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value="Fungi").update(archived=True) + with self.assertRaisesRegex(ValueError, "archived"): + tagging_api.update_tag_in_taxonomy(self.taxonomy, "Fungi", "Mushrooms") + assert get_tag("Fungi").value == "Fungi" + + def test_update_tag_in_taxonomy_archived_taxonomy(self) -> None: + Taxonomy.objects.filter(pk=self.taxonomy.pk).update(archived=True) + self.taxonomy.refresh_from_db() + with self.assertRaisesRegex(ValueError, "archived"): + tagging_api.update_tag_in_taxonomy(self.taxonomy, "Fungi", "Mushrooms") + assert get_tag("Fungi").value == "Fungi" + def test_copy_tags(self) -> None: obj1 = "object_id1" obj2 = "object_id2" @@ -1094,3 +1327,18 @@ def test_depth_limit(self) -> None: tag_4 = tagging_api.add_tag_to_taxonomy(taxonomy, "Patty - depth 5", parent_tag_value=tag_4.value) with pytest.raises(ValidationError, match="Cannot create tags more than 6 levels deep."): tagging_api.add_tag_to_taxonomy(taxonomy, "Deep - depth 6", parent_tag_value=tag_4.value) + + def test_add_tag_to_taxonomy_archived_taxonomy(self) -> None: + Taxonomy.objects.filter(pk=self.taxonomy.pk).update(archived=True) + self.taxonomy.refresh_from_db() + tag_count = self.taxonomy.tag_set.count() + with self.assertRaisesRegex(ValueError, "archived"): + tagging_api.add_tag_to_taxonomy(self.taxonomy, "New tag") + assert self.taxonomy.tag_set.count() == tag_count + + def test_add_tag_to_taxonomy_archived_parent(self) -> None: + Tag.objects.filter(taxonomy=self.taxonomy, value="Animalia").update(archived=True) + tag_count = self.taxonomy.tag_set.count() + with self.assertRaisesRegex(ValueError, "archived"): + tagging_api.add_tag_to_taxonomy(self.taxonomy, "New tag", parent_tag_value="Animalia") + assert self.taxonomy.tag_set.count() == tag_count diff --git a/tests/openedx_tagging/test_migrations.py b/tests/openedx_tagging/test_migrations.py index ca1c5c906..7dd59e2c9 100644 --- a/tests/openedx_tagging/test_migrations.py +++ b/tests/openedx_tagging/test_migrations.py @@ -1,5 +1,5 @@ """ -Tests for the 0022_tag_external_id_not_null migration's backfill logic. +Tests that run openedx_tagging migrations over rows created on the earlier schema. Uses the `migrator` pytest fixture from django_test_migrations to run the actual migration against historical (frozen) model states, the same way it will run against @@ -12,6 +12,9 @@ MIGRATE_FROM = ("oel_tagging", "0021_remove_system_defined_add_read_only") MIGRATE_TO = ("oel_tagging", "0022_tag_external_id_not_null") +MIGRATE_0023_FROM = ("oel_tagging", "0022_tag_external_id_not_null") +MIGRATE_0023_TO = ("oel_tagging", "0023_archived_and_deletion_locked") + @pytest.mark.django_db def test_backfill_generates_missing_external_ids(migrator) -> None: @@ -98,3 +101,34 @@ def test_reverse_migration_is_a_noop_that_keeps_data(migrator) -> None: reverted_tag = RevertedTag.objects.get(pk=tag.pk) assert reverted_tag.external_id == "Rollback Tag" + + +@pytest.mark.django_db +def test_archived_and_deletion_locked_default_to_false_for_existing_rows(migrator) -> None: + """ + Rows that exist before 0023 come out unarchived and unlocked. + """ + old_state = migrator.apply_initial_migration(MIGRATE_0023_FROM) + Taxonomy = old_state.apps.get_model("oel_tagging", "Taxonomy") + Tag = old_state.apps.get_model("oel_tagging", "Tag") + ObjectTag = old_state.apps.get_model("oel_tagging", "ObjectTag") + + taxonomy = Taxonomy.objects.create(name="Flags Test", export_id="flags_test") + tag = Tag.objects.create( + taxonomy=taxonomy, value="Flags Tag", external_id="flags-tag", depth=0, lineage="Flags Tag\t", + ) + object_tag = ObjectTag.objects.create(object_id="object:id:1", taxonomy=taxonomy, tag=tag) + + new_state = migrator.apply_tested_migration(MIGRATE_0023_TO) + NewTaxonomy = new_state.apps.get_model("oel_tagging", "Taxonomy") + NewTag = new_state.apps.get_model("oel_tagging", "Tag") + NewObjectTag = new_state.apps.get_model("oel_tagging", "ObjectTag") + + new_taxonomy = NewTaxonomy.objects.get(pk=taxonomy.pk) + new_tag = NewTag.objects.get(pk=tag.pk) + new_object_tag = NewObjectTag.objects.get(pk=object_tag.pk) + + assert new_taxonomy.archived is False + assert new_tag.archived is False + assert new_object_tag.archived is False + assert new_object_tag.deletion_locked is False diff --git a/tests/openedx_tagging/test_models.py b/tests/openedx_tagging/test_models.py index abcdc9728..1031d67f5 100644 --- a/tests/openedx_tagging/test_models.py +++ b/tests/openedx_tagging/test_models.py @@ -613,6 +613,13 @@ def test_cast(self): == " object:id:1: Life on Earth=Bacteria" ) + def test_copying_a_tag_via_casting_copies_field_values(self) -> None: + self.object_tag.archived = True + self.object_tag.deletion_locked = True + copy_tag = ObjectTagTestSubclass.cast(self.object_tag) + assert copy_tag.archived is True + assert copy_tag.deletion_locked is True + def test_object_tag_export_id(self): # ObjectTag's export_id defaults to its taxonomy's export_id assert self.object_tag.export_id == self.taxonomy.export_id @@ -1072,6 +1079,32 @@ def test_rename(self): assert self.bob.depth == 1 assert self.bob.lineage == "Charlie\tBob\t" + def test_child_and_descendant_counts_exclude_archived(self): + assert self.charlie.child_count == 2 + assert self.charlie.descendant_count == 5 + + Tag.objects.filter(pk__in=[self.bob.pk, self.foxtrot.pk]).update(archived=True) + charlie = Tag.objects.get(pk=self.charlie.pk) # The counts are cached on the instance + assert charlie.child_count == 1 # Bob + assert charlie.descendant_count == 3 # Bob and Foxtrot + + def test_save_taxonomy_less_tag_does_not_cascade_to_other_taxonomy_less_tags(self): + """ + Tags without a taxonomy have no tree of their own, so renaming one must not rewrite the lineage of other + taxonomy-less tags that happen to share its old lineage prefix. + """ + # `external_id` is set because save() can only generate one from a taxonomy. + tag = Tag.objects.create(taxonomy=None, value="Orphan", external_id="orphan") + other_root = Tag.objects.create(taxonomy=None, value="Orphan", external_id="other_orphan") + other_child = Tag.objects.create(taxonomy=None, value="Child", external_id="other_child", parent=other_root) + assert other_child.lineage == "Orphan\tChild\t" + + tag.value = "Renamed" + tag.save() + + other_child.refresh_from_db() + assert other_child.lineage == "Orphan\tChild\t" + # TODO: The following event-emission tests don't really belong in TestTagLineage. # They should be moved to a separate test_events.py module. @patch("openedx_tagging.signal_handlers.emit_content_object_associations_changed_for_tag_task.delay") diff --git a/tests/openedx_tagging/test_views.py b/tests/openedx_tagging/test_views.py index 8848f513c..38528ba8f 100644 --- a/tests/openedx_tagging/test_views.py +++ b/tests/openedx_tagging/test_views.py @@ -246,6 +246,27 @@ def test_list_taxonomy_query_count(self): assert not taxonomy["can_delete_taxonomy"] assert not taxonomy["can_tag_object"] + def test_list_taxonomy_excludes_archived(self) -> None: + api.create_taxonomy(name="Active taxonomy") + archived = api.create_taxonomy(name="Archived taxonomy") + Taxonomy.objects.filter(pk=archived.pk).update(archived=True) + + self.client.force_authenticate(user=self.staff) + response = self.client.get(TAXONOMY_LIST_URL) + assert response.status_code == status.HTTP_200_OK + assert [t["name"] for t in response.data["results"]] == ["Active taxonomy"] + + def test_list_taxonomy_tags_count_excludes_archived(self) -> None: + taxonomy = api.create_taxonomy(name="Taxonomy with archived tags") + for i in range(3): + Tag.objects.create(taxonomy=taxonomy, value=f"Tag {i}") + Tag.objects.filter(taxonomy=taxonomy, value="Tag 0").update(archived=True) + + self.client.force_authenticate(user=self.staff) + response = self.client.get(TAXONOMY_LIST_URL) + assert response.status_code == status.HTTP_200_OK + assert response.data["results"][0]["tags_count"] == 2 + def test_list_invalid_page(self) -> None: url = TAXONOMY_LIST_URL @@ -300,6 +321,28 @@ def test_detail_read_only_taxonomy(self): assert response.status_code == status.HTTP_200_OK assert response.data["read_only"] is True + def test_get_archived_taxonomy_responds_404(self) -> None: + taxonomy = api.create_taxonomy(name="Archived taxonomy") + Taxonomy.objects.filter(pk=taxonomy.pk).update(archived=True) + + self.client.force_authenticate(user=self.staff) + response = self.client.get(TAXONOMY_DETAIL_URL.format(pk=taxonomy.pk)) + assert response.status_code == status.HTTP_404_NOT_FOUND + + def test_get_taxonomy_tags_count_excludes_archived(self) -> None: + """ + Only the list view annotates `tags_count`, so this covers the serializer's fallback count. + """ + taxonomy = api.create_taxonomy(name="Taxonomy with archived tags") + for i in range(3): + Tag.objects.create(taxonomy=taxonomy, value=f"Tag {i}") + Tag.objects.filter(taxonomy=taxonomy, value="Tag 0").update(archived=True) + + self.client.force_authenticate(user=self.staff) + response = self.client.get(TAXONOMY_DETAIL_URL.format(pk=taxonomy.pk)) + assert response.status_code == status.HTTP_200_OK + assert response.data["tags_count"] == 2 + def test_detail_taxonomy_404(self) -> None: url = TAXONOMY_DETAIL_URL.format(pk=123123) @@ -491,6 +534,17 @@ def test_update_taxonomy_read_only(self, read_only): assert response.data["name"] == "new name" assert response.data["read_only"] == (not read_only) + def test_update_archived_taxonomy_is_400(self): + taxonomy = api.create_taxonomy(name="Archived taxonomy") + Taxonomy.objects.filter(pk=taxonomy.pk).update(archived=True) + + self.client.force_authenticate(user=self.staff) + response = self.client.put(TAXONOMY_DETAIL_URL.format(pk=taxonomy.pk), {"name": "new name"}, format="json") + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "This taxonomy is archived and cannot be edited." in str(response.data) + taxonomy.refresh_from_db() + assert taxonomy.name == "Archived taxonomy" + def test_update_taxonomy_404(self): url = TAXONOMY_DETAIL_URL.format(pk=123123) @@ -549,6 +603,17 @@ def test_patch_taxonomy_read_only(self, read_only): assert response.status_code == status.HTTP_200_OK assert response.data["read_only"] == (not read_only) + def test_patch_archived_taxonomy_is_400(self): + taxonomy = api.create_taxonomy(name="Archived taxonomy") + Taxonomy.objects.filter(pk=taxonomy.pk).update(archived=True) + + self.client.force_authenticate(user=self.staff) + response = self.client.patch(TAXONOMY_DETAIL_URL.format(pk=taxonomy.pk), {"name": "new name"}, format="json") + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "This taxonomy is archived and cannot be edited." in str(response.data) + taxonomy.refresh_from_db() + assert taxonomy.name == "Archived taxonomy" + def test_patch_taxonomy_404(self): url = TAXONOMY_DETAIL_URL.format(pk=123123) @@ -579,6 +644,18 @@ def test_delete_taxonomy(self, user_attr, expected_status): response = self.client.get(url) assert response.status_code == status.HTTP_404_NOT_FOUND + def test_delete_archived_taxonomy(self): + """ + Deleting an archived taxonomy still works. Archived deletion rules are owned by a separate ticket (#780). + """ + taxonomy = api.create_taxonomy(name="Archived taxonomy") + Taxonomy.objects.filter(pk=taxonomy.pk).update(archived=True) + + self.client.force_authenticate(user=self.staff) + response = self.client.delete(TAXONOMY_DETAIL_URL.format(pk=taxonomy.pk)) + assert response.status_code == status.HTTP_204_NO_CONTENT + assert not Taxonomy.objects.filter(pk=taxonomy.pk).exists() + def test_delete_taxonomy_404(self): url = TAXONOMY_DETAIL_URL.format(pk=123123) @@ -586,6 +663,15 @@ def test_delete_taxonomy_404(self): response = self.client.delete(url) assert response.status_code == status.HTTP_404_NOT_FOUND + def test_export_archived_taxonomy_responds_404(self): + taxonomy = api.create_taxonomy(name="Archived taxonomy") + api.add_tag_to_taxonomy(taxonomy=taxonomy, tag="Tag 1") + Taxonomy.objects.filter(pk=taxonomy.pk).update(archived=True) + + self.client.force_authenticate(user=self.staff) + response = self.client.get(TAXONOMY_EXPORT_URL.format(pk=taxonomy.pk), {"output_format": "json"}) + assert response.status_code == status.HTTP_404_NOT_FOUND + @ddt.data( ("csv", "text"), ("json", "application/json") @@ -881,6 +967,17 @@ def test_retrieve_object_tags_query_count(self): assert response.data[object_id]["taxonomies"][0]["can_tag_object"] == expected_perm assert response.data[object_id]["taxonomies"][0]["tags"] == sort_test_applied_result + def test_retrieve_object_tags_excludes_archived(self): + object_id = "abc" + api.tag_object(object_id=object_id, taxonomy=self.taxonomy, tags=["Mammalia", "Fungi"]) + ObjectTag.objects.filter(object_id=object_id, _value="Fungi").update(archived=True) + + self.client.force_authenticate(user=self.staff) + response = self.client.get(OBJECT_TAGS_RETRIEVE_URL.format(object_id=object_id)) + assert response.status_code == status.HTTP_200_OK + tags = response.data[object_id]["taxonomies"][0]["tags"] + assert [tag["value"] for tag in tags] == ["Mammalia"] + def test_retrieve_object_tags_unauthorized(self): """ Test retrieving object tags from an unauthorized object_id @@ -1320,6 +1417,16 @@ def check(object_id_pattern: str, count_implicit=False): "course07-unit02-problem02.2": 2, } + def test_get_counts_excludes_archived(self): + api.tag_object(object_id="course01-problem01", taxonomy=self.free_text_taxonomy, tags=["a", "b"]) + ObjectTag.objects.filter(object_id="course01-problem01", _value="b").update(archived=True) + + url = OBJECT_TAG_COUNTS_URL.format(object_id_pattern="course01-*") + with self.assertNumQueries(1): + result = self.client.get(url) + assert result.status_code == status.HTTP_200_OK + assert result.data == {"course01-problem01": 1} + def test_get_counts_invalid_spec(self): """ Test handling of an invalid object tag pattern @@ -1791,6 +1898,47 @@ def test_large_search(self): " Tag 119 (Tag 118) (children: 0)", ] + def test_list_excludes_archived_tag(self): + self.client.force_authenticate(user=self.staff) + Tag.objects.filter(taxonomy=self.small_taxonomy, value="Fungi").update(archived=True) + + response = self.client.get(f"{self.small_taxonomy_url}?parent_tag=Eukaryota") + assert response.status_code == status.HTTP_200_OK + values = [tag["value"] for tag in response.data["results"]] + assert "Fungi" not in values + assert "Plantae" in values + + def test_list_archived_parent_tag_404(self): + self.client.force_authenticate(user=self.staff) + Tag.objects.filter(taxonomy=self.small_taxonomy, value="Chordata").update(archived=True) + + response = self.client.get(f"{self.small_taxonomy_url}?parent_tag=Chordata") + assert response.status_code == status.HTTP_404_NOT_FOUND + response = self.client.get(f"{self.small_taxonomy_url}?parent_tag=Chordata&full_depth_threshold=1000") + assert response.status_code == status.HTTP_404_NOT_FOUND + + def test_list_full_depth_excludes_archived_descendant(self): + self.client.force_authenticate(user=self.staff) + Tag.objects.filter(taxonomy=self.small_taxonomy, value="Mammalia").update(archived=True) + + response = self.client.get(f"{self.small_taxonomy_url}?parent_tag=Animalia&full_depth_threshold=1000") + assert response.status_code == status.HTTP_200_OK + values = [tag["value"] for tag in response.data["results"]] + assert "Chordata" in values + assert "Mammalia" not in values + + def test_list_archived_taxonomy_404(self): + self.client.force_authenticate(user=self.staff) + Taxonomy.objects.filter(pk=self.small_taxonomy.pk).update(archived=True) + assert not self.small_taxonomy.tag_set.filter(archived=True).exists() # Only the taxonomy is archived + + response = self.client.get(self.small_taxonomy_url) + assert response.status_code == status.HTTP_404_NOT_FOUND + response = self.client.get(f"{self.small_taxonomy_url}?search_term=eu") + assert response.status_code == status.HTTP_404_NOT_FOUND + response = self.client.get(f"{self.small_taxonomy_url}?parent_tag=Eukaryota") + assert response.status_code == status.HTTP_404_NOT_FOUND + def test_get_children(self): self._build_large_taxonomy() self.client.force_authenticate(user=self.staff) @@ -2037,6 +2185,58 @@ def test_create_tag_in_taxonomy_with_duplicate_external_id(self): # Neither rejected request should have created a new tag: assert self.small_taxonomy.tag_set.count() == tag_count_before + def test_create_tag_with_archived_value_returns_archived_message(self): + self.client.force_authenticate(user=self.staff) + Tag.objects.filter(taxonomy=self.small_taxonomy, value="Archaea").update(archived=True) + tag_count_before = self.small_taxonomy.tag_set.count() + + response = self.client.post(self.small_taxonomy_url, {"tag": "Archaea"}, format="json") + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data == { + "tag": ['Tag value "Archaea" belongs to an archived tag in this taxonomy. Use a different value.'] + } + assert self.small_taxonomy.tag_set.count() == tag_count_before + + def test_create_tag_with_archived_external_id_returns_archived_message(self): + self.client.force_authenticate(user=self.staff) + archaea = self.small_taxonomy.tag_set.get(value="Archaea") + Tag.objects.filter(pk=archaea.pk).update(archived=True) + + response = self.client.post( + self.small_taxonomy_url, {"tag": "New Tag", "external_id": archaea.external_id}, format="json", + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data == [ + f"Tag external_id '{archaea.external_id}' belongs to an archived tag in this taxonomy. " + "Use a different external_id." + ] + assert not self.small_taxonomy.tag_set.filter(value="New Tag").exists() + + def test_create_tag_with_unarchived_duplicate_value_keeps_message(self): + self.client.force_authenticate(user=self.staff) + + response = self.client.post(self.small_taxonomy_url, {"tag": "Archaea"}, format="json") + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data == {"tag": ['Tag value "Archaea" already exists in this taxonomy.']} + + def test_update_tag_to_archived_value_returns_archived_message(self): + self.client.force_authenticate(user=self.staff) + Tag.objects.filter(taxonomy=self.small_taxonomy, value="Archaea").update(archived=True) + + response = self.client.put( + self.small_taxonomy_url, + {"tag": "Bacteria", "updated_tag_value": "Archaea"}, + format="json", + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.data["updated_tag_value"] == [ + 'Tag value "Archaea" belongs to an archived tag in this taxonomy. Use a different value.' + ] + def test_create_tag_in_invalid_taxonomy(self): self.client.force_authenticate(user=self.staff) new_tag_value = "New Tag" @@ -2423,6 +2623,75 @@ def test_update_tag_in_taxonomy_reflects_changes_in_object_tags(self): object_tag_3.refresh_from_db() self.assertEqual(object_tag_3.value, updated_tag_value) + def test_update_tag_in_taxonomy_archived_tag_is_400(self): + self.client.force_authenticate(user=self.staff) + fungi = Tag.objects.get(taxonomy=self.small_taxonomy, value="Fungi") + Tag.objects.filter(pk=fungi.pk).update(archived=True) + + response = self.client.put( + self.small_taxonomy_url, {"tag": "Fungi", "updated_tag_value": "Mushrooms"}, format="json" + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + fungi.refresh_from_db() + assert fungi.value == "Fungi" + + def test_update_tag_in_taxonomy_archived_taxonomy_is_400(self): + self.client.force_authenticate(user=self.staff) + fungi = Tag.objects.get(taxonomy=self.small_taxonomy, value="Fungi") + Taxonomy.objects.filter(pk=self.small_taxonomy.pk).update(archived=True) # The tag itself stays unarchived + + update_data = {"tag": "Fungi", "updated_tag_value": "Mushrooms"} + for method in (self.client.put, self.client.patch): + response = method(self.small_taxonomy_url, update_data, format="json") + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "archived" in response.content.decode() + fungi.refresh_from_db() + assert fungi.value == "Fungi" + + def test_create_tag_in_archived_taxonomy_is_400(self): + self.client.force_authenticate(user=self.staff) + Taxonomy.objects.filter(pk=self.small_taxonomy.pk).update(archived=True) + tag_count = self.small_taxonomy.tag_set.count() + + response = self.client.post(self.small_taxonomy_url, {"tag": "New Tag"}, format="json") + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "archived" in response.content.decode() + assert self.small_taxonomy.tag_set.count() == tag_count + assert not self.small_taxonomy.tag_set.filter(value="New Tag").exists() + + def test_create_tag_under_archived_parent_is_400(self): + self.client.force_authenticate(user=self.staff) + Tag.objects.filter(taxonomy=self.small_taxonomy, value="Animalia").update(archived=True) + + response = self.client.post( + self.small_taxonomy_url, {"tag": "New Tag", "parent_tag_value": "Animalia"}, format="json" + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "archived" in response.content.decode() + assert not self.small_taxonomy.tag_set.filter(value="New Tag").exists() + + def test_update_missing_tag_in_archived_taxonomy_is_400(self): + self.client.force_authenticate(user=self.staff) + Taxonomy.objects.filter(pk=self.small_taxonomy.pk).update(archived=True) + + response = self.client.put( + self.small_taxonomy_url, {"tag": "No such tag", "updated_tag_value": "Anything"}, format="json" + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "archived" in response.content.decode() + + def test_delete_tag_in_archived_taxonomy_is_404(self): + """ + Deleting tags uses the archive-aware lookup. Archived deletion rules are owned by a separate ticket (#780). + """ + self.client.force_authenticate(user=self.staff) + Taxonomy.objects.filter(pk=self.small_taxonomy.pk).update(archived=True) + + response = self.client.delete(self.small_taxonomy_url, {"tags": ["Fungi"], "with_subtags": True}, format="json") + assert response.status_code == status.HTTP_404_NOT_FOUND + assert self.small_taxonomy.tag_set.filter(value="Fungi").exists() + def test_update_tag_in_taxonomy_with_invalid_tag(self): self.client.force_authenticate(user=self.staff) updated_tag_value = "Updated Tag" @@ -3467,6 +3736,20 @@ def test_import_plan(self, file_format: str) -> None: self._check_taxonomy_not_changed() + def test_import_archived_taxonomy_is_400(self) -> None: + """ + Both the import and its plan are rejected for an archived taxonomy, and its tags stay as they were. + """ + Taxonomy.objects.filter(pk=self.taxonomy.pk).update(archived=True) + self.client.force_authenticate(user=self.staff) + + for url in (TAXONOMY_TAGS_IMPORT_URL, TAXONOMY_TAGS_IMPORT_PLAN_URL): + file = self._get_file([{"id": "tag_1", "value": "Tag 1"}], "json") + response = self.client.put(url.format(pk=self.taxonomy.id), {"file": file}, format="multipart") + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "This taxonomy is archived and cannot be edited." in str(response.data) + assert set(self.taxonomy.tag_set.values_list("value", flat=True)) == {"Old tag 1", "Old tag 2"} + def test_import_no_file(self) -> None: """ Tests importing a taxonomy without a file.