feat: add ci

fastapi-admin
Pierre-Louis Guhur 1 year ago
parent d9dce84767
commit c8044acd75

@ -0,0 +1,29 @@
name: test-driven development
on: [push]
jobs:
build:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.10", "3.11"]
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-dev.txt
- name: Static typing with mypy
run: |
# stop the build if there are MyPy errors
mypy app
- name: Test with pytest
run: |
SQLITE=True pytest

@ -1,4 +1,5 @@
from collections import defaultdict
from typing import DefaultDict
from sqlalchemy.orm import Session
from sqlalchemy import func
from majority_judgment import majority_judgment
@ -127,7 +128,7 @@ def create_vote(db: Session, vote: schemas.VoteCreate) -> schemas.VoteGet:
return db_vote
def get_vote(db: Session, vote_id: int) -> schemas.VoteGet:
def get_vote(db: Session, vote_id: int) -> models.Vote:
# TODO check with JWT tokens the authorization
votes_by_id = db.query(models.Vote).filter(models.Vote.id == vote_id)
@ -136,8 +137,9 @@ def get_vote(db: Session, vote_id: int) -> schemas.VoteGet:
"votes", f"Several votes have the same primary keys {vote_id}"
)
if votes_by_id.count() == 1:
return votes_by_id.first()
vote = votes_by_id.first()
if vote is not None:
return vote
votes_by_ref = db.query(models.Vote).filter(models.Vote.ref == vote_id)
@ -146,8 +148,9 @@ def get_vote(db: Session, vote_id: int) -> schemas.VoteGet:
"votes", f"Several votes have the same reference {vote_id}"
)
if votes_by_ref.count() == 1:
return votes_by_ref.first()
vote = votes_by_ref.first()
if vote is not None:
return vote
raise errors.NotFoundError("votes")
@ -164,7 +167,7 @@ def get_results(db: Session, election_id: int) -> schemas.ResultsGet:
.group_by(models.Vote.candidate_id, models.Vote.grade_id)
.all()
)
ballots = defaultdict(dict)
ballots: DefaultDict[int, dict[int, int]] = defaultdict(dict)
for candidate_id, grade_value, num_votes in db_votes:
ballots[candidate_id][grade_value] = num_votes
merit_profile = {

@ -1,23 +1,24 @@
from __future__ import annotations
from urllib.parse import quote
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from .settings import settings
# database_url = (
# "postgresql+psycopg2://"
# f"{settings.postgres_name}:{quote(settings.postgres_password)}"
# f"@{settings.postgres_host}:{settings.postgres_port}"
# f"/{settings.postgres_db}"
# )
# engine = create_engine(database_url)
if settings.sqlite:
database_url = "sqlite:///./main.db"
engine = create_engine(database_url, connect_args={"check_same_thread": False})
database_url = "sqlite:///./main.db"
engine = create_engine(
database_url, connect_args={"check_same_thread": False}
)
else:
database_url = (
"postgresql+psycopg2://"
f"{settings.postgres_name}:{quote(settings.postgres_password)}"
f"@{settings.postgres_host}:{settings.postgres_port}"
f"/{settings.postgres_db}"
)
engine = create_engine(database_url)
SessionLocal: sessionmaker = sessionmaker(
SessionLocal: sessionmaker = sessionmaker( # type: ignore
autocommit=False, autoflush=False, bind=engine
)

@ -5,13 +5,6 @@ from pydantic.fields import ModelField
from .settings import settings
def _empty_string():
"""
Using the default factory for field
"""
return ""
class ArgumentsSchemaError(Exception):
"""
An error occured on the arguments provided to a schema
@ -19,9 +12,9 @@ class ArgumentsSchemaError(Exception):
Name = t.Annotated[str, Field(min_length=1, max_length=255)]
Ref = t.Annotated[str, Field(default_factory=_empty_string, max_length=255)]
Image = t.Annotated[str, Field(default_factory=_empty_string, max_length=255)]
Description = t.Annotated[str, Field(default_factory=_empty_string, max_length=1024)]
Ref = t.Annotated[str, Field(..., max_length=255)]
Image = t.Annotated[str, Field(..., max_length=255)]
Description = t.Annotated[str, Field(..., max_length=1024)]
Color = t.Annotated[str, Field(min_length=3, max_length=10)]
@ -58,8 +51,8 @@ def _causal_dates_validator(*fields: str):
class CandidateBase(BaseModel):
name: Name
description: Description
image: Image
description: Description = ""
image: Image = ""
date_created: datetime = Field(default_factory=datetime.now)
date_modified: datetime = Field(default_factory=datetime.now)
@ -80,7 +73,7 @@ class CandidateGet(CandidateBase):
class GradeBase(BaseModel):
name: Name
value: int = Field(ge=0, lt=settings.max_grades, pre=True)
description: Description
description: Description = ""
date_created: datetime = Field(default_factory=datetime.now)
date_modified: datetime = Field(default_factory=datetime.now)
@ -137,8 +130,8 @@ def _in_a_long_time() -> datetime:
class ElectionBase(BaseModel):
name: Name
description: Description
ref: Ref
description: Description = ""
ref: Ref = ""
date_created: datetime = Field(default_factory=datetime.now)
date_modified: datetime = Field(default_factory=datetime.now)
num_voters: int = Field(0, ge=0, le=settings.max_voters)

@ -2,6 +2,8 @@ from pydantic import BaseSettings
class Settings(BaseSettings):
sqlite: bool = False
postgres_password: str = ""
postgres_db: str = "mj"
postgres_name: str = "mj"

@ -5,15 +5,15 @@ import random
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from ..database import Base
from ..database import Base, get_db
from .. import schemas
from ..main import app, get_db
from ..main import app
test_database_url = "sqlite:///./test.db"
test_engine = create_engine(
test_database_url, connect_args={"check_same_thread": False}
)
TestingSessionLocal: sessionmaker = sessionmaker(
TestingSessionLocal: sessionmaker = sessionmaker( # type: ignore
autocommit=False, autoflush=False, bind=test_engine
)
@ -58,8 +58,10 @@ def _random_election(num_candidates: int, num_grades: int) -> RandomElection:
"""
Generate an election with random names
"""
grades = [dict(name=_random_string(10), value=i) for i in range(num_grades)]
candidates = [dict(name=_random_string(10)) for i in range(num_candidates)]
grades: list[dict[str, int | str]] = [
{"name": _random_string(10), "value": i} for i in range(num_grades)
]
candidates = [{"name": _random_string(10)} for i in range(num_candidates)]
name = _random_string(10)
return {"candidates": candidates, "grades": grades, "name": name}

@ -24,14 +24,14 @@ def test_grade_default_values():
assert grade.date_modified >= grade.date_created
# Automatic conversion helps to load data from the payload
grade = GradeBase(name="foo", value=1.2, description="bar foo")
grade = GradeBase(name="foo", value=1.2, description="bar foo") # type: ignore
assert grade.value == 1
grade = GradeBase(name="foo", value="1", description="bar foo")
grade = GradeBase(name="foo", value="1", description="bar foo") # type: ignore
assert grade.value == 1
# Any field name is accepted
grade = GradeBase(name="foo", value=1, foo="bar")
grade = GradeBase(name="foo", value=1, foo="bar") # type: ignore
def test_grade_validation_value():
@ -40,7 +40,7 @@ def test_grade_validation_value():
"""
with pytest.raises(ValidationError):
# value must be a positive integer
GradeBase(name="foo", value="bar", description="bar foo")
GradeBase(name="foo", value="bar", description="bar foo") # type: ignore
with pytest.raises(ValidationError):
GradeBase(name="foo", value=-1, description="bar foo")

@ -1,22 +0,0 @@
from django.contrib import admin
from .models import Election, Vote, Token
admin.site.register(Vote)
admin.site.register(Token)
class ElectionAdmin(admin.ModelAdmin):
list_display = (
"title",
"candidates",
"on_invitation_only",
"num_grades",
"start_at",
"finish_at",
"select_language",
"restrict_results",
)
admin.site.register(Election, ElectionAdmin)

@ -1,5 +0,0 @@
from django.apps import AppConfig
class ElectionConfig(AppConfig):
name = "election"

@ -1,53 +0,0 @@
# Generated by Django 3.0.3 on 2020-04-16 14:45
import django.contrib.postgres.fields
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='Election',
fields=[
('id', models.CharField(db_index=True, max_length=20, primary_key=True, serialize=False)),
('title', models.CharField(max_length=255, verbose_name='Title')),
('candidates', django.contrib.postgres.fields.ArrayField(base_field=models.CharField(max_length=255, verbose_name='Name'), size=None)),
('on_invitation_only', models.BooleanField(default=False)),
('num_grades', models.PositiveSmallIntegerField(verbose_name='Num. grades')),
('start_at', models.IntegerField(default=1587048322, verbose_name='Start date')),
('finish_at', models.IntegerField(default=1587048323, verbose_name='End date')),
('select_language', models.CharField(default='en', max_length=2, verbose_name='Language')),
('restrict_results', models.BooleanField(default=True)),
],
options={
'abstract': False,
},
),
migrations.CreateModel(
name='Vote',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('grades_by_candidate', django.contrib.postgres.fields.ArrayField(base_field=models.SmallIntegerField(verbose_name='Note'), size=None)),
('election', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='election.Election')),
],
),
migrations.CreateModel(
name='Token',
fields=[
('id', models.CharField(db_index=True, max_length=20, primary_key=True, serialize=False)),
('email', models.EmailField(max_length=254)),
('used', models.BooleanField(default=False, verbose_name='Used')),
('election', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='election.Election')),
],
options={
'abstract': False,
},
),
]

@ -1,29 +0,0 @@
# Generated by Django 3.0.3 on 2020-04-28 10:52
from django.db import migrations, models
import time
class Migration(migrations.Migration):
dependencies = [
('election', '0001_initial'),
]
operations = [
migrations.AlterField(
model_name='election',
name='finish_at',
field=models.IntegerField(default=time.time, verbose_name='End date'),
),
migrations.AlterField(
model_name='election',
name='restrict_results',
field=models.BooleanField(default=False),
),
migrations.AlterField(
model_name='election',
name='start_at',
field=models.IntegerField(default=time.time, verbose_name='Start date'),
),
]

@ -1,17 +0,0 @@
# Generated by Django 3.0.3 on 2020-05-10 09:31
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('election', '0002_auto_20200428_1052'),
]
operations = [
migrations.RemoveField(
model_name='token',
name='email',
),
]

@ -1,18 +0,0 @@
# Generated by Django 3.1.1 on 2021-04-10 15:13
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('election', '0003_remove_token_email'),
]
operations = [
migrations.AddField(
model_name='election',
name='send_mail',
field=models.BooleanField(default=True),
),
]

@ -1,85 +0,0 @@
import logging
from time import time
from django.contrib.postgres.fields import ArrayField
from django.db import IntegrityError, models
from django.conf import settings
from libs.django_randomprimary import RandomPrimaryIdModel
logger = logging.getLogger(__name__)
class Election(RandomPrimaryIdModel):
title = models.CharField("Title", max_length=255)
candidates = ArrayField(models.CharField("Name", max_length=255))
on_invitation_only = models.BooleanField(default=False)
num_grades = models.PositiveSmallIntegerField("Num. grades", null=False)
start_at = models.IntegerField("Start date", default=time)
finish_at = models.IntegerField("End date", default=time)
send_mail = models.BooleanField(default=True)
# Language preference is used for emailing voters
select_language = models.CharField("Language", max_length=2, default="en")
# If results are restricted, one can see them only when the election is finished
restrict_results = models.BooleanField(default=False)
# add some constraints before saving the database
def save(self, *args, **kwargs):
# make sure we don't ask for more grades than allowed in the database
if self.num_grades is None:
raise IntegrityError("Election requires a positive number of grades.")
if self.num_grades > settings.MAX_NUM_GRADES or self.num_grades <= 0:
raise IntegrityError(
"Max number of grades is %d. Asked for %d grades"
% (self.num_grades, settings.MAX_NUM_GRADES)
)
# check that the title is not empty
if self.title is None or self.title == "":
raise IntegrityError("Election requires a proper title")
# check that the language is known
if not self.select_language in settings.LANGUAGE_AVAILABLE:
raise IntegrityError(
"Election is only available in " + settings.LANGUAGE_AVAILABLE
)
# check if the end date is not in the past
if self.finish_at <= round(time()):
raise IntegrityError("The election cannot be over in the past")
# check if the end date is not before the begin date
if self.start_at > self.finish_at:
raise IntegrityError("The election can't end until it has started")
return super().save(*args, **kwargs)
class Vote(models.Model):
election = models.ForeignKey(Election, on_delete=models.CASCADE)
grades_by_candidate = ArrayField(models.SmallIntegerField("Note"))
def save(self, *args, **kwargs):
logging.debug(self.grades_by_candidate, self.election.candidates)
if len(self.grades_by_candidate) != len(self.election.candidates):
raise IntegrityError(
"number of grades (%d) differs from number of candidates (%d)"
% (len(self.grades_by_candidate), len(self.election.candidates))
)
if not all(
0 <= mention < self.election.num_grades
for mention in self.grades_by_candidate
):
raise IntegrityError(
"grades have to be between 0 and %d" % (self.election.num_grades - 1)
)
return super().save(*args, **kwargs)
class Token(RandomPrimaryIdModel):
election = models.ForeignKey(Election, on_delete=models.CASCADE)
used = models.BooleanField("Used", default=False)

@ -1,112 +0,0 @@
from django.utils.text import slugify
from rest_framework import serializers
from election.models import Election, Vote, Token
from django.conf import settings
class ElectionCreateSerializer(serializers.ModelSerializer):
elector_emails = serializers.ListField(
child=serializers.EmailField(),
write_only=True,
required=False,
)
def create(self, data):
# Copy the validated_data
validated_data = dict(data)
validated_data["on_invitation_only"] = False
if "elector_emails" in validated_data:
if validated_data["elector_emails"] != []:
validated_data["on_invitation_only"] = True
validated_data.pop("elector_emails")
return Election.objects.create(**validated_data)
class Meta:
model = Election
fields = (
"title",
"candidates",
"on_invitation_only",
"num_grades",
"elector_emails",
"start_at",
"finish_at",
"select_language",
"restrict_results",
"send_mail",
)
def to_representation(self, instance):
ret = super().to_representation(instance)
ret["slug"] = slugify(instance.title)
ret["id"] = instance.id
ret["tokens"] = [token.id for token in Token.objects.filter(election=instance)]
return ret
class ElectionViewSerializer(serializers.ModelSerializer):
class Meta:
model = Election
fields = "__all__"
def to_representation(self, instance):
ret = super().to_representation(instance)
ret["slug"] = slugify(instance.title)
ret["id"] = instance.id
return ret
class VoteSerializer(serializers.ModelSerializer):
grades_by_candidate = serializers.ListField(
child=serializers.IntegerField(
min_value=0,
max_value=settings.MAX_NUM_GRADES - 1,
)
)
token = serializers.CharField(write_only=True, required=False)
def create(self, validated_data):
# Copy the validated_data
validated_data = dict(validated_data)
try:
validated_data.pop("token")
except KeyError:
pass
return Vote.objects.create(**validated_data)
class Meta:
model = Vote
fields = (
"grades_by_candidate",
"election",
"token",
)
# See https://github.com/MieuxVoter/mvapi/pull/5#discussion_r291891403 for explanations
class Candidate:
def __init__(self, name, idx, profile, grade):
self.name = name
self.id = idx
self.profile = profile
self.grade = grade
class CandidateSerializer(serializers.Serializer):
name = serializers.CharField()
id = serializers.IntegerField(min_value=0)
profile = serializers.DictField(child=serializers.IntegerField())
grade = serializers.IntegerField(min_value=0, max_value=settings.MAX_NUM_GRADES)
class LinkSerializer(serializers.Serializer):
election_id = serializers.CharField()
select_language = serializers.CharField(max_length=2, required=False)
emails = serializers.ListField(
child=serializers.EmailField(), write_only=True, required=True
)

@ -1,215 +0,0 @@
import logging
from django.core import mail
from django.db import IntegrityError
from django.test import TestCase
from rest_framework.test import APITestCase
import election.urls as urls
from election.models import MAX_NUM_GRADES, Election, Token, Vote
from libs.majority_judgment import majority_judgment, compute_votes
# To avoid undesirable logging messages due to 400 Error.
logger = logging.getLogger("django.request")
logger.setLevel(logging.ERROR)
class ElectionCreateAPIViewTestCase(APITestCase):
def test_create_election(self):
title = "Super รฉlection - utf-8 chars: ๐Ÿคจ ๐Ÿ˜ ๐Ÿ˜‘ ๐Ÿ˜ถ ๐Ÿ™„ ๐Ÿ˜ ๐Ÿ˜ฃ ๐Ÿ˜ฅ ๐Ÿ˜ฎ ๐Ÿค ๐Ÿ˜ฏ ๐Ÿ˜ช ๐Ÿ˜ซ ๐Ÿ˜ด ๐Ÿ˜Œ ๐Ÿ˜› ๐Ÿ˜œ ๐Ÿ˜ ๐Ÿคค ๐Ÿ˜’ ๐Ÿ˜“ ๐Ÿ˜” ๐Ÿ˜• ๐Ÿ™ƒ ๐Ÿค‘ ๐Ÿ˜ฒ โ˜น๏ธ ๐Ÿ™ ๐Ÿ˜– ๐Ÿ˜ž ๐Ÿ˜Ÿ ๐Ÿ˜ค ๐Ÿ˜ข ๐Ÿ˜ญ ๐Ÿ˜ฆ ๐Ÿ˜ง ๐Ÿ˜จ ๐Ÿ˜ฉ ๐Ÿคฏ !"
candidates = [
"Seb",
"Pierre-Louis",
]
response_post = self.client.post(
urls.new_election(),
{
"title": title,
"candidates": candidates,
"on_invitation_only": False,
"num_grades": 5,
},
)
self.assertEqual(201, response_post.status_code)
election_pk = response_post.data["id"]
response_get = self.client.get(urls.election_details(election_pk))
self.assertEqual(200, response_get.status_code)
self.assertEqual(title, response_get.data["title"])
self.assertEqual(candidates, response_get.data["candidates"])
def test_mandatory_fields(self):
# Missing num_grades
self.assertRaises(
IntegrityError,
Election.objects.create,
candidates=["Seb", "PL"],
title="My election",
)
# Missing candidates
self.assertRaises(
IntegrityError, Election.objects.create, num_grades=5, title="My election"
)
# Missing title
self.assertRaises(
IntegrityError,
Election.objects.create,
candidates=["Seb", "PL"],
num_grades=5,
)
class VoteOnInvitationViewTestCase(APITestCase):
def setUp(self):
self.election = Election.objects.create(
title="Test election",
candidates=[
"Seb",
"Pierre-Louis",
],
on_invitation_only=True,
num_grades=5,
)
self.token = Token.objects.create(
election=self.election,
email="joe@example.com",
)
def test_valid_vote(self):
response = self.client.post(
urls.vote(),
{
"election": self.election.id,
"grades_by_candidate": [0, 0],
"token": self.token.id,
},
)
self.assertEqual(201, response.status_code)
def test_vote_without_token(self):
response = self.client.post(
urls.vote(),
{
"election": self.election.id,
"grades_by_candidate": [0, 0],
},
)
self.assertEqual(400, response.status_code)
def test_vote_wrong_token(self):
response = self.client.post(
urls.vote(),
{
"election": self.election.id,
"grades_by_candidate": [0, 0],
# make sure the token is not the good one
"token": self.token.id + "#abc",
},
)
self.assertEqual(400, response.status_code)
def test_vote_already_used_token(self):
for _ in range(2):
response = self.client.post(
urls.vote(),
{
"election": self.election.id,
"grades_by_candidate": [0, 0],
"token": self.token.id,
},
)
self.assertEqual(400, response.status_code)
class MailForCreationTestCase(TestCase):
def test_send_mail(self):
response_post = self.client.post(
urls.new_election(),
{
"title": "Test",
"candidates": ["A", "B"],
"on_invitation_only": False,
"num_grades": 5,
"elector_emails": ["name@example.com"],
},
)
election_pk = response_post.data["id"]
self.assertEqual(len(mail.outbox), 1)
self.assertIn("/vote/" + election_pk, mail.outbox[0].body)
class ResutsTestCase(TestCase):
def setUp(self):
self.election = Election.objects.create(
title="Test election",
candidates=[
"Seb",
"Pierre-Louis",
],
num_grades=5,
on_invitation_only=True,
)
self.votes = [
Vote.objects.create(election=self.election, grades_by_candidate=[1, 2]),
Vote.objects.create(election=self.election, grades_by_candidate=[1, 2]),
Vote.objects.create(election=self.election, grades_by_candidate=[1, 3]),
Vote.objects.create(election=self.election, grades_by_candidate=[2, 1]),
]
self.election_no_vote = Election.objects.create(
title="Election without votes", candidates=["Clement", "Seb"], num_grades=7
)
def test_results_with_majority_judgment(self):
profiles, scores, grades = compute_votes(
[v.grades_by_candidate for v in self.votes], self.election.num_grades
)
sorted_indexes = majority_judgment(profiles)
assert sorted_indexes == [1, 0]
def test_num_grades(self):
self.assertRaises(
IntegrityError,
Vote.objects.create,
election=self.election,
grades_by_candidate=[1, 6],
)
def test_view_existing_election(self):
response = self.client.get(urls.results(self.election.id))
self.assertEqual(200, response.status_code)
"""def test_ongoing_election(self):
self.election.is_finished = False
self.election.restrict_results = False
self.election.save()
response = self.client.get(
urls.results(self.election.id)
)
self.assertEqual(400, response.status_code)
self.election.is_finished = True
self.election.restrict_results = True
self.election.save()"""
def test_opened_election_without_vote(self):
response = self.client.get(urls.results(self.election_no_vote.id))
self.assertEqual(400, response.status_code)
def test_opened_election_with_vote(self):
response = self.client.get(urls.results(self.election.id))
self.assertEqual(200, response.status_code)

@ -1,33 +0,0 @@
from django.urls import path, reverse
from election.views import *
app_name = "election"
urlpatterns = [
path(r"", ElectionCreateAPIView.as_view(), name="create"),
path(r"get/<str:pk>/", ElectionDetailsAPIView.as_view(), name="details"),
path(r"vote/", VoteAPIView.as_view(), name="vote"),
path(r"results/<str:pk>/", ResultAPIView.as_view(), name="results"),
path(r"links/", LinkAPIView.as_view(), name="links"),
]
def new_election():
return reverse("election:create")
def election_details(election_pk):
return reverse("election:details", args=(election_pk,))
def vote():
return reverse("election:vote")
def results(election_pk):
return reverse("election:results", args=(election_pk,))
def links():
return reverse("election:links")

@ -1,361 +0,0 @@
import os
import urllib
import base64
from typing import Optional, Dict, Tuple, List
from time import time
from django.db import IntegrityError
from django.conf import settings
from django.core.mail import EmailMultiAlternatives
from django.template.loader import render_to_string
from django.utils.translation import activate, gettext
from rest_framework import status
from rest_framework.decorators import api_view
from rest_framework.generics import CreateAPIView, RetrieveAPIView
from rest_framework.response import Response
from rest_framework.request import Request
from rest_framework.views import APIView
import election.serializers as serializers
from election.models import Election, Token, Vote
from libs import majority_judgment as mj
# Error codes:
UNKNOWN_ELECTION_ERROR = "E1: Unknown election"
ONGOING_ELECTION_ERROR = "E2: Ongoing election"
NO_VOTE_ERROR = "E3: No recorded vote"
ELECTION_NOT_STARTED_ERROR = "E4: Election not started"
ELECTION_FINISHED_ERROR = "E5: Election finished"
INVITATION_ONLY_ERROR = "E6: Election on invitation only, please provide token"
UNKNOWN_TOKEN_ERROR = "E7: Wrong token"
USED_TOKEN_ERROR = "E8: Token already used"
WRONG_ELECTION_ERROR = "E9: Parameters for the election are incorrect"
SEND_MAIL_ERROR = "E10: Error sending email"
# A Grade is always given a int
Grade = int
def send_mails_invitation_api(list_email_token: list, election: str):
"""
Def to send the election invitation by API
"""
for couple in list_email_token:
token_get: str = f"?token={couple[1]}"
merge_data: Dict[str, str] = {
"invitation_url": f"{settings.SITE_URL}/vote/{election.id}{token_get}",
"result_url": f"{settings.SITE_URL}/result/{election.id}",
"title": election.title,
}
if election.select_language not in settings.LANGUAGE_AVAILABLE:
activate(settings.DEFAULT_LANGUAGE)
else:
activate(election.select_language)
text_body = render_to_string("election/mail_invitation.txt", merge_data)
html_body = render_to_string("election/mail_invitation.html", merge_data)
data = urllib.parse.urlencode(
{
"from": "Mieux Voter <" + settings.DEFAULT_FROM_EMAIL + ">",
"to": couple[0],
"subject": f"[{gettext('Mieux Voter')}] {election.title}",
"text": text_body,
"html": html_body,
"o:tracking": False,
"o:tag": "Invitation",
"o:require-tls": settings.EMAIL_USE_TLS,
"o:skip-verification": settings.EMAIL_SKIP_VERIFICATION,
},
doseq=True,
).encode()
send_api(data)
def send_mail_api(email: str, text_body, html_body, title):
"""
Def to send mails by API
"""
data = urllib.parse.urlencode(
{
"from": "Mieux Voter <" + settings.DEFAULT_FROM_EMAIL + ">",
"to": email,
"subject": f"[{gettext('Mieux Voter')}] {title}",
"text": text_body,
"html": html_body,
"o:tracking": False,
"o:tag": "Invitation",
"o:require-tls": settings.EMAIL_USE_TLS,
"o:skip-verification": settings.EMAIL_SKIP_VERIFICATION,
},
doseq=True,
).encode()
send_api(data)
def send_api(data):
"""
def to do api request
"""
request = urllib.request.Request(settings.EMAIL_API_DOMAIN, data=data)
encoded_token = base64.b64encode(
("api:" + settings.EMAIL_API_KEY).encode("ascii")
).decode("ascii")
request.add_header("Authorization", "Basic {}".format(encoded_token))
try:
urllib.request.urlopen(request)
except Exception as err:
return err
def send_mails_invitation_smtp(list_email_token: list, election: str):
"""
Def to send the election invitation by SMTP
"""
for couple in list_email_token:
token_get: str = f"?token={couple[1]}"
merge_data: Dict[str, str] = {
"invitation_url": f"{settings.SITE_URL}/vote/{election.id}{token_get}",
"result_url": f"{settings.SITE_URL}/result/{election.id}",
"title": election.title,
}
if election.select_language not in settings.LANGUAGE_AVAILABLE:
activate(settings.DEFAULT_LANGUAGE)
else:
activate(election.select_language)
text_body = render_to_string("election/mail_invitation.txt", merge_data)
html_body = render_to_string("election/mail_invitation.html", merge_data)
msg = EmailMultiAlternatives(
f"[{gettext('Mieux Voter')}] {election.title}",
text_body,
settings.EMAIL_HOST_USER,
[couple[0]],
)
msg.attach_alternative(html_body, "text/html")
msg.send()
class ElectionCreateAPIView(CreateAPIView):
serializer_class = serializers.ElectionCreateSerializer
def create(self, request: Request, *args, **kwargs) -> Response:
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
election = serializer.save()
electors_emails = serializer.validated_data.get("elector_emails", [])
list_email_token = []
for email in electors_emails:
token = Token.objects.create(
election=election,
)
list_email_token.append([email, token.id])
if election.send_mail:
if settings.EMAIL_TYPE == "API":
send_mails_invitation_api(list_email_token, election)
else:
send_mails_invitation_smtp(list_email_token, election)
headers = self.get_success_headers(serializer.data)
return Response(
serializer.data, status=status.HTTP_201_CREATED, headers=headers
)
class ElectionDetailsAPIView(RetrieveAPIView):
serializer_class = serializers.ElectionViewSerializer
def get(self, request: Request, pk: str, **kwargs) -> Response:
try:
election = Election.objects.get(id=pk)
except Election.DoesNotExist:
return Response(
UNKNOWN_ELECTION_ERROR,
status=status.HTTP_400_BAD_REQUEST,
)
if round(time()) < election.start_at:
return Response(
ELECTION_NOT_STARTED_ERROR,
status=status.HTTP_401_UNAUTHORIZED,
)
serializer = serializers.ElectionViewSerializer(election)
return Response(
serializer.data,
status=status.HTTP_200_OK,
)
class VoteAPIView(CreateAPIView):
"""
View to vote in an election
"""
serializer_class = serializers.VoteSerializer
def create(self, request: Request, *args, **kwargs) -> Response:
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
election = serializer.validated_data["election"]
if round(time()) >= election.finish_at:
return Response(
ELECTION_FINISHED_ERROR,
status=status.HTTP_401_UNAUTHORIZED,
)
if election.on_invitation_only:
try:
token = serializer.validated_data["token"]
except KeyError:
return Response(
INVITATION_ONLY_ERROR,
status=status.HTTP_400_BAD_REQUEST,
)
try:
token_object = Token.objects.get(
election=election,
id=token,
)
except Token.DoesNotExist:
return Response(
UNKNOWN_TOKEN_ERROR,
status=status.HTTP_400_BAD_REQUEST,
)
if token_object.used:
return Response(
USED_TOKEN_ERROR,
status=status.HTTP_400_BAD_REQUEST,
)
token_object.used = True
token_object.save()
# Dealing with potential errors like the number of mentions
# differs from the number of candidates.
try:
self.perform_create(serializer)
except IntegrityError:
return Response(WRONG_ELECTION_ERROR, status=status.HTTP_400_BAD_REQUEST)
headers = self.get_success_headers(serializer.data)
return Response(status=status.HTTP_201_CREATED, headers=headers)
class ResultAPIView(APIView):
"""
View to list the result of an election using majority judgment.
"""
def get(self, request, pk, **kwargs):
try:
election = Election.objects.get(id=pk)
except Election.DoesNotExist:
return Response(
UNKNOWN_ELECTION_ERROR,
status=status.HTTP_400_BAD_REQUEST,
)
except IntegrityError:
return Response(
WRONG_ELECTION_ERROR,
status=status.HTTP_400_BAD_REQUEST,
)
if election.restrict_results and round(time()) < election.finish_at:
return Response(
ONGOING_ELECTION_ERROR,
status=status.HTTP_400_BAD_REQUEST,
)
votes = Vote.objects.filter(election=election)
if len(votes) == 0:
return Response(
NO_VOTE_ERROR,
status=status.HTTP_400_BAD_REQUEST,
)
votes: List[List[Grade]] = [v.grades_by_candidate for v in votes]
merit_profiles: List[Dict[Grade, int]] = mj.votes_to_merit_profiles(
votes, range(election.num_grades)
)
indexed_values: List[
Tuple[int, mj.MajorityValue]
] = mj.sort_by_value_with_index(
[mj.MajorityValue(profil) for profil in merit_profiles]
)
print(len(indexed_values))
candidates = [
serializers.Candidate(
election.candidates[idx],
idx,
merit_profiles[idx],
value.grade,
)
for idx, value in indexed_values
]
serializer = serializers.CandidateSerializer(candidates, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
class LinkAPIView(CreateAPIView):
"""
View to send the result and vote links if it is an open election
"""
serializer_class = serializers.LinkSerializer
def create(self, request: Request, *args, **kwargs) -> Response:
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
election_id = serializer.validated_data["election_id"]
select_language = serializer.validated_data["select_language"]
try:
election = Election.objects.get(id=election_id)
except Election.DoesNotExist:
return Response(
WRONG_ELECTION_ERROR,
status=status.HTTP_400_BAD_REQUEST,
)
emails = serializer.validated_data.get("emails", [])
merge_data: Dict[str, str] = {
"result_url": f"{settings.SITE_URL}/result/{election.id}",
"title": election.title,
}
if (
select_language == None
or select_language not in settings.LANGUAGE_AVAILABLE
):
select_language = election.select_language
activate(select_language)
if election.on_invitation_only:
text_body = render_to_string("election/mail_one_link.txt", merge_data)
html_body = render_to_string("election/mail_one_link.html", merge_data)
else:
merge_data["vote_url"] = f"{settings.SITE_URL}/vote/{election.id}"
text_body = render_to_string("election/mail_two_links.txt", merge_data)
html_body = render_to_string("election/mail_two_links.html", merge_data)
send_status = send_mail_api(emails, text_body, html_body, election.title)
return Response(status=send_status)

@ -1,15 +0,0 @@
#!/usr/bin/env python
import os
import sys
if __name__ == "__main__":
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mvapi.settings")
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
raise ImportError(
"Couldn't import Django. Are you sure it's installed and "
"available on your PYTHONPATH environment variable? Did you "
"forget to activate a virtual environment?"
) from exc
execute_from_command_line(sys.argv)

@ -1,160 +0,0 @@
"""
Django settings for mvapi project.
For more information on this file, see
https://docs.djangoproject.com/en/2.1/topics/settings/
For the full list of settings and their values, see
https://docs.djangoproject.com/en/2.1/ref/settings/
"""
import os
from django.http.request import RAISE_ERROR
# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SITE_URL = os.environ["SITE_URL"]
DEBUG = os.environ["DJANGO_DEBUG"] == "True"
SECRET_KEY = os.environ["DJANGO_SECRET_KEY"]
ALLOWED_HOSTS = os.environ["DJANGO_ALLOWED_HOSTS"].split(",")
MAX_NUM_GRADES = int(os.environ["MAX_NUM_GRADES"])
DEFAULT_LANGUAGE = "en"
LANGUAGE_AVAILABLE = os.environ.get("LANGUAGE_AVAILABLE", DEFAULT_LANGUAGE)
# Application definition
INSTALLED_APPS = [
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.messages",
"django.contrib.staticfiles",
"django_extensions",
"dbbackup", # django-dbbackup
"rest_framework",
"corsheaders",
"election",
]
DBBACKUP_STORAGE = "django.core.files.storage.FileSystemStorage"
DBBACKUP_STORAGE_OPTIONS = {"location": "backup"}
MIDDLEWARE = [
"corsheaders.middleware.CorsMiddleware",
"django.middleware.security.SecurityMiddleware",
"django.contrib.sessions.middleware.SessionMiddleware",
"django.middleware.common.CommonMiddleware",
"django.middleware.csrf.CsrfViewMiddleware",
"django.contrib.auth.middleware.AuthenticationMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
"django.middleware.clickjacking.XFrameOptionsMiddleware",
"django.middleware.locale.LocaleMiddleware",
]
CORS_ORIGIN_ALLOW_ALL = True
ROOT_URLCONF = "mvapi.urls"
TEMPLATES = [
{
"BACKEND": "django.template.backends.django.DjangoTemplates",
"DIRS": [os.path.join(BASE_DIR, "templates")],
"APP_DIRS": True,
"OPTIONS": {
"context_processors": [
"django.template.context_processors.i18n",
"django.template.context_processors.debug",
"django.template.context_processors.request",
"django.contrib.auth.context_processors.auth",
"django.contrib.messages.context_processors.messages",
],
},
},
]
WSGI_APPLICATION = "mvapi.wsgi.application"
# Database
# https://docs.djangoproject.com/en/2.1/ref/settings/#databases
DATABASES = {
"default": {
"ENGINE": "django.db.backends.postgresql",
"NAME": "postgres",
"USER": "postgres",
"HOST": "db",
"PORT": 5432,
}
}
# Password validation
# https://docs.djangoproject.com/en/2.1/ref/settings/#auth-password-validators
AUTH_PASSWORD_VALIDATORS = [
{
"NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator",
},
{
"NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",
},
{
"NAME": "django.contrib.auth.password_validation.CommonPasswordValidator",
},
{
"NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",
},
]
# Internationalization
# https://docs.djangoproject.com/en/2.1/topics/i18n/
LANGUAGE_CODE = "en-us"
TIME_ZONE = "UTC"
USE_I18N = True
USE_L10N = True
USE_TZ = True
LOCALE_PATHS = (os.path.join(BASE_DIR, "locale"),)
# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/2.1/howto/static-files/
STATIC_URL = "/static/"
################################################################################
# #
# MAIL SETTINGS #
# #
################################################################################
if os.environ["EMAIL_USE_TLS"] in ("True", "true", "on", "1"):
EMAIL_USE_TLS = True
else:
EMAIL_USE_TLS = False
EMAIL_TYPE = os.environ["EMAIL_TYPE"]
if EMAIL_TYPE == "API":
# To use the Mailgun's API
EMAIL_API_KEY = os.environ["EMAIL_API_KEY"]