diff --git a/dnscms/dnscms/management/__init__.py b/dnscms/dnscms/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dnscms/dnscms/management/commands/__init__.py b/dnscms/dnscms/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dnscms/dnscms/management/commands/galtinn_users.py b/dnscms/dnscms/management/commands/galtinn_users.py new file mode 100644 index 0000000..991f992 --- /dev/null +++ b/dnscms/dnscms/management/commands/galtinn_users.py @@ -0,0 +1,93 @@ +import csv + +from django.contrib.auth.models import User +from django.core.management.base import BaseCommand, CommandError +from django.db import transaction + +from dnscms.models import GaltinnUser + +FIELDS = [ + "cms_id", + "username", + "email", + "first_name", + "last_name", + "is_superuser", + "is_active", + "last_login", + "galtinn_id", +] + + +class Command(BaseCommand): + help = "Export local users as CSV, or link them to Galtinn users from a reviewed CSV" + + def add_arguments(self, parser): + subparsers = parser.add_subparsers(dest="action", required=True) + subparsers.add_parser("export", help="Write all users as CSV to stdout") + link = subparsers.add_parser( + "link", help="Link users to the galtinn_id given per cms_id; blank rows are skipped" + ) + link.add_argument("csv_file") + link.add_argument("--apply", action="store_true", help="Save the links (default: dry run)") + + def handle(self, *args, **options): + if options["action"] == "export": + self.export() + else: + self.link(options["csv_file"], apply=options["apply"]) + + def export(self): + writer = csv.DictWriter(self.stdout, fieldnames=FIELDS, lineterminator="\n") + writer.writeheader() + for user in User.objects.select_related("galtinn_user").order_by("pk"): + galtinn_user = getattr(user, "galtinn_user", None) + writer.writerow( + { + "cms_id": user.pk, + "username": user.username, + "email": user.email, + "first_name": user.first_name, + "last_name": user.last_name, + "is_superuser": int(user.is_superuser), + "is_active": int(user.is_active), + "last_login": user.last_login.date().isoformat() if user.last_login else "", + "galtinn_id": galtinn_user.galtinn_id if galtinn_user else "", + } + ) + + def link(self, csv_file, *, apply): + with open(csv_file, newline="") as f: + rows = [row for row in csv.DictReader(f) if (row.get("galtinn_id") or "").strip()] + + users = User.objects.in_bulk([row["cms_id"] for row in rows]) + current = {link.user_id: link.galtinn_id for link in GaltinnUser.objects.all()} + owners = {galtinn_id: user_id for user_id, galtinn_id in current.items()} + errors, links = [], [] + for row in rows: + cms_id, galtinn_id = int(row["cms_id"]), row["galtinn_id"].strip() + user = users.get(cms_id) + if user is None: + errors.append(f"user #{cms_id} does not exist") + elif user.username != row["username"]: + errors.append(f"user #{cms_id} is {user.username!r}, not {row['username']!r}") + elif current.get(cms_id) == galtinn_id: + continue + elif cms_id in current: + errors.append(f"{user.username!r} is already linked to Galtinn #{current[cms_id]}") + elif galtinn_id in owners: + errors.append(f"Galtinn #{galtinn_id} is given to more than one user") + else: + owners[galtinn_id] = cms_id + links.append(GaltinnUser(user=user, galtinn_id=galtinn_id)) + if errors: + raise CommandError("Nothing linked:\n" + "\n".join(errors)) + + for link in links: + self.stdout.write(f"{link.user.username} -> Galtinn #{link.galtinn_id}") + if apply: + with transaction.atomic(): + GaltinnUser.objects.bulk_create(links) + self.stdout.write(f"Linked {len(links)} users") + else: + self.stdout.write(f"Dry run: would link {len(links)} users, rerun with --apply") diff --git a/dnscms/tests/test_galtinn_users.py b/dnscms/tests/test_galtinn_users.py new file mode 100644 index 0000000..a6cd5d5 --- /dev/null +++ b/dnscms/tests/test_galtinn_users.py @@ -0,0 +1,117 @@ +import csv +import io + +import pytest +from django.contrib.auth.models import User +from django.core.management import CommandError, call_command + +from dnscms.models import GaltinnUser + +pytestmark = pytest.mark.django_db + + +def write_csv(tmp_path, rows): + path = tmp_path / "users.csv" + with path.open("w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=["cms_id", "username", "galtinn_id"]) + writer.writeheader() + writer.writerows(rows) + return str(path) + + +def run(*args): + out = io.StringIO() + call_command("galtinn_users", *args, stdout=out) + return out.getvalue() + + +@pytest.fixture +def kari(): + return User.objects.create_user(username="kari", email="kari@neuf.test", password="pw") + + +@pytest.fixture +def ola(): + return User.objects.create_superuser(username="ola", email="ola@neuf.test", password="pw") + + +def test_export_lists_users_with_existing_links(kari, ola): + GaltinnUser.objects.create(user=kari, galtinn_id="42") + + rows = list(csv.DictReader(io.StringIO(run("export")))) + + assert [(row["username"], row["galtinn_id"], row["is_superuser"]) for row in rows] == [ + ("kari", "42", "0"), + ("ola", "", "1"), + ] + assert rows[0]["cms_id"] == str(kari.pk) + assert rows[0]["email"] == "kari@neuf.test" + + +def test_link_is_a_dry_run_by_default(tmp_path, kari): + path = write_csv(tmp_path, [{"cms_id": kari.pk, "username": "kari", "galtinn_id": "42"}]) + + output = run("link", path) + + assert "kari -> Galtinn #42" in output + assert "Dry run" in output + assert not GaltinnUser.objects.exists() + + +def test_link_apply_links_filled_rows_and_keeps_passwords(tmp_path, kari, ola): + path = write_csv( + tmp_path, + [ + {"cms_id": kari.pk, "username": "kari", "galtinn_id": " 42 "}, + {"cms_id": ola.pk, "username": "ola", "galtinn_id": ""}, + ], + ) + + run("link", path, "--apply") + + assert GaltinnUser.objects.get().galtinn_id == "42" + assert kari.galtinn_user.galtinn_id == "42" + kari.refresh_from_db() + assert kari.has_usable_password() + + +def test_link_apply_is_repeatable(tmp_path, kari): + path = write_csv(tmp_path, [{"cms_id": kari.pk, "username": "kari", "galtinn_id": "42"}]) + + run("link", path, "--apply") + output = run("link", path, "--apply") + + assert "Linked 0 users" in output + assert GaltinnUser.objects.count() == 1 + + +@pytest.mark.parametrize( + ("rows", "message"), + [ + ([{"cms_id": 999, "username": "ghost", "galtinn_id": "1"}], "does not exist"), + ( + [{"cms_id": "KARI", "username": "someone-else", "galtinn_id": "1"}], + "not 'someone-else'", + ), + ([{"cms_id": "KARI", "username": "kari", "galtinn_id": "7"}], "already linked"), + ( + [ + {"cms_id": "OLA", "username": "ola", "galtinn_id": "1"}, + {"cms_id": "OLA2", "username": "ola2", "galtinn_id": "1"}, + ], + "more than one user", + ), + ([{"cms_id": "OLA", "username": "ola", "galtinn_id": "42"}], "more than one user"), + ], +) +def test_link_refuses_the_whole_file_on_any_bad_row(tmp_path, kari, ola, rows, message): + GaltinnUser.objects.create(user=kari, galtinn_id="42") + ola2 = User.objects.create_user(username="ola2") + ids = {"KARI": kari.pk, "OLA": ola.pk, "OLA2": ola2.pk} + good = {"cms_id": ola2.pk, "username": "ola2", "galtinn_id": "500"} + rows = [{**row, "cms_id": ids.get(row["cms_id"], row["cms_id"])} for row in rows] + + with pytest.raises(CommandError, match=message): + run("link", write_csv(tmp_path, [*rows, good]), "--apply") + + assert GaltinnUser.objects.count() == 1