dnscms: add galtinn_users command to export and link local users

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