Skip to content

Commit

Permalink
Ajout commande pour exporter contacts lié à une structure avec email …
Browse files Browse the repository at this point in the history
…vide + commande pour maj email
  • Loading branch information
alanzirek committed Nov 15, 2024
1 parent 514bd52 commit 8735a63
Show file tree
Hide file tree
Showing 4 changed files with 112 additions and 0 deletions.
18 changes: 18 additions & 0 deletions core/management/commands/export_contacts_structures.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
from django.core.management.base import BaseCommand
import csv
import sys

from core.models import Structure


class Command(BaseCommand):
help = """
Exporte tous les contacts reliés à une structure.
Usage: python manage.py export_contacts_structures > contacts_structures.csv
"""

def handle(self, *args, **options):
writer = csv.writer(sys.stdout, delimiter=";")
writer.writerow(["structure (niveau1)", "structure (niveau2)", "email"])
for structure in Structure.objects.filter(contact__email=""):
writer.writerow([structure.niveau1, structure.niveau2, ""])
31 changes: 31 additions & 0 deletions core/management/commands/update_contacts_structures_email.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
from django.core.management.base import BaseCommand
import csv

from core.models import Contact


class Command(BaseCommand):
help = """
Ajoute les emails des contacts liés à une structure qui n'ont pas d'email renseigné.
Les adresses emails sont récupérées depuis le fichier CSV fourni en argument (fichier généré par la commande export_contacts_structures).
Usage: python manage.py update_contacts_emails contacts_structures.csv
"""

def add_arguments(self, parser):
parser.add_argument("csv_file", type=str)

def handle(self, *args, **options):
updates = 0
with open(options["csv_file"], newline="") as f:
reader = csv.DictReader(f, delimiter=";")
for row in reader:
niveau1 = row["structure (niveau1)"]
niveau2 = row["structure (niveau2)"]
email = row["email"].strip()

if email:
updates += Contact.objects.filter(
structure__niveau1=niveau1, structure__niveau2=niveau2, email=""
).update(email=email)

self.stdout.write(f"Mise à jour effectuée : {updates} contacts modifiés")
21 changes: 21 additions & 0 deletions core/tests/test_export_contacts_structures.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
import pytest
from io import StringIO
from django.core.management import call_command
from core.models import Structure, Contact
from contextlib import redirect_stdout


@pytest.mark.django_db
def test_export_contacts_with_data():
structure = Structure.objects.create(niveau1="Direction", niveau2="Service A")
Contact.objects.create(structure=structure, email="")
Contact.objects.create(structure=structure, email="[email protected]")

out = StringIO()
with redirect_stdout(out):
call_command("export_contacts_structures")

result = out.getvalue().splitlines()
assert len(result) == 2
assert result[0] == "structure (niveau1);structure (niveau2);email"
assert result[1] == "Direction;Service A;"
42 changes: 42 additions & 0 deletions core/tests/test_update_contacts_structures_email.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
import pytest
from io import StringIO
from contextlib import redirect_stdout
from django.core.management import call_command
from core.models import Structure, Contact


@pytest.fixture
def mock_csv_data(tmp_path):
data = (
"structure (niveau1);structure (niveau2);email\n"
"Direction;Service A;[email protected]\n"
"Direction;Service B;[email protected]\n"
"Direction;Service C;"
)
p = tmp_path / "test.csv"
p.write_text(data, encoding="utf-8")
return str(p)


@pytest.mark.django_db
def test_update_contacts_emails(mock_csv_data):
structure_a = Structure.objects.create(niveau1="Direction", niveau2="Service A")
structure_b = Structure.objects.create(niveau1="Direction", niveau2="Service B")
structure_c = Structure.objects.create(niveau1="Direction", niveau2="Service C")

contact_a = Contact.objects.create(structure=structure_a, email="")
contact_b = Contact.objects.create(structure=structure_b, email="")
contact_c = Contact.objects.create(structure=structure_c, email="")

out = StringIO()
with redirect_stdout(out):
call_command("update_contacts_structures_email", mock_csv_data)

assert "Mise à jour effectuée : 2 contacts modifiés" in out.getvalue()

contact_a.refresh_from_db()
contact_b.refresh_from_db()
contact_c.refresh_from_db()
assert contact_a.email == "[email protected]"
assert contact_b.email == "[email protected]"
assert contact_c.email == ""

0 comments on commit 8735a63

Please sign in to comment.