Mirror
/
app-MAIL-temp
mirror of https://github.com/simple-login/app.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #381 from TheLastProject/feature/importExportTests 

Feature/import export tests
This commit is contained in:
Son Nguyen Kim 2021-03-15 15:17:58 +01:00 committed by GitHub
parent cf6442cec2 e02d95216f
commit ca4d097f14
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 331 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
app/api/views/export.py
View File

@ -55,7 +55,14 @@ def export_aliases():
data = [["alias", "note", "enabled", "mailboxes"]]
for alias in Alias.filter_by(user_id=user.id).all(): # type: Alias
mailboxes = " ".join([mailbox.email for mailbox in alias.mailboxes])
# Always put the main mailbox first
# It is seen a primary while importing
alias_mailboxes = alias.mailboxes
alias_mailboxes.insert(
0, alias_mailboxes.pop(alias_mailboxes.index(alias.mailbox))
)
mailboxes = " ".join([mailbox.email for mailbox in alias_mailboxes])
data.append([alias.email, alias.note, alias.enabled, mailboxes])
si = StringIO()

102
app/import_utils.py Normal file
View File

@ -0,0 +1,102 @@
from .log import LOG
import csv
import requests
from app import s3
from app.email_utils import get_email_domain_part
from app.extensions import db
from app.models import (
Alias,
AliasMailbox,
BatchImport,
CustomDomain,
DeletedAlias,
DomainDeletedAlias,
Mailbox,
User,
)
from app.utils import sanitize_email
def handle_batch_import(batch_import: BatchImport):
user = batch_import.user
batch_import.processed = True
db.session.commit()
LOG.debug("Start batch import for %s %s", batch_import, user)
file_url = s3.get_url(batch_import.file.path)
LOG.d("Download file %s from %s", batch_import.file, file_url)
r = requests.get(file_url)
lines = [line.decode() for line in r.iter_lines()]
import_from_csv(user, lines)
def import_from_csv(batch_import: BatchImport, user: User, lines):
reader = csv.DictReader(lines)
for row in reader:
try:
full_alias = sanitize_email(row["alias"])
note = row["note"]
except KeyError:
LOG.warning("Cannot parse row %s", row)
continue
alias_domain = get_email_domain_part(full_alias)
custom_domain = CustomDomain.get_by(domain=alias_domain)
if (
not custom_domain
or not custom_domain.verified
or custom_domain.user_id != user.id
):
LOG.debug("domain %s can't be used %s", alias_domain, user)
continue
if (
Alias.get_by(email=full_alias)
or DeletedAlias.get_by(email=full_alias)
or DomainDeletedAlias.get_by(email=full_alias)
):
LOG.d("alias already used %s", full_alias)
continue
mailboxes = []
if "mailboxes" in row:
for mailbox_email in row["mailboxes"].split():
mailbox_email = sanitize_email(mailbox_email)
mailbox = Mailbox.get_by(email=mailbox_email)
if not mailbox or not mailbox.verified or mailbox.user_id != user.id:
LOG.d("mailbox %s can't be used %s", mailbox, user)
continue
mailboxes.append(mailbox.id)
if len(mailboxes) == 0:
mailboxes = [user.default_mailbox_id]
alias = Alias.create(
user_id=user.id,
email=full_alias,
note=note,
mailbox_id=mailboxes[0],
custom_domain_id=custom_domain.id,
batch_import_id=batch_import.id,
)
db.session.commit()
db.session.flush()
LOG.d("Create %s", alias)
for i in range(1, len(mailboxes)):
alias_mailbox = AliasMailbox.create(
alias_id=alias.id,
mailbox_id=mailboxes[i],
)
db.session.commit()
LOG.d("Create %s", alias_mailbox)

96
job_runner.py
View File

@ -2,13 +2,10 @@
Run scheduled jobs.
Not meant for running job at precise time (+- 1h)
"""
import csv
import time
import arrow
import requests
from app import s3
from app.config import (
JOB_ONBOARDING_1,
JOB_ONBOARDING_2,
@ -18,22 +15,11 @@ from app.config import (
from app.email_utils import (
send_email,
render,
get_email_domain_part,
)
from app.utils import sanitize_email
from app.import_utils import handle_batch_import
from app.extensions import db
from app.log import LOG
from app.models import (
User,
Job,
BatchImport,
Alias,
DeletedAlias,
DomainDeletedAlias,
CustomDomain,
Mailbox,
AliasMailbox,
)
from app.models import User, Job, BatchImport
from server import create_app
@ -113,84 +99,6 @@ def onboarding_mailbox(user):
)
def handle_batch_import(batch_import: BatchImport):
user = batch_import.user
batch_import.processed = True
db.session.commit()
LOG.debug("Start batch import for %s %s", batch_import, user)
file_url = s3.get_url(batch_import.file.path)
LOG.d("Download file %s from %s", batch_import.file, file_url)
r = requests.get(file_url)
lines = [line.decode() for line in r.iter_lines()]
reader = csv.DictReader(lines)
for row in reader:
try:
full_alias = sanitize_email(row["alias"])
note = row["note"]
except KeyError:
LOG.warning("Cannot parse row %s", row)
continue
alias_domain = get_email_domain_part(full_alias)
custom_domain = CustomDomain.get_by(domain=alias_domain)
if (
not custom_domain
or not custom_domain.verified
or custom_domain.user_id != user.id
):
LOG.debug("domain %s can't be used %s", alias_domain, user)
continue
if (
Alias.get_by(email=full_alias)
or DeletedAlias.get_by(email=full_alias)
or DomainDeletedAlias.get_by(email=full_alias)
):
LOG.d("alias already used %s", full_alias)
continue
mailboxes = []
if "mailboxes" in row:
for mailbox_email in row["mailboxes"].split():
mailbox_email = sanitize_email(mailbox_email)
mailbox = Mailbox.get_by(email=mailbox_email)
if not mailbox or not mailbox.verified or mailbox.user_id != user.id:
LOG.d("mailbox %s can't be used %s", mailbox, user)
continue
mailboxes.append(mailbox.id)
if len(mailboxes) == 0:
mailboxes = [user.default_mailbox_id]
alias = Alias.create(
user_id=user.id,
email=full_alias,
note=note,
mailbox_id=mailboxes[0],
custom_domain_id=custom_domain.id,
batch_import_id=batch_import.id,
)
db.session.commit()
db.session.flush()
LOG.d("Create %s", alias)
for i in range(1, len(mailboxes)):
alias_mailbox = AliasMailbox.create(
alias_id=alias.id,
mailbox_id=mailboxes[i],
)
db.session.commit()
LOG.d("Create %s", alias_mailbox)
if __name__ == "__main__":
while True:
# run a job 1h earlier or later is not a big deal ...

4
static/batch_import_template.csv vendored
View File

@ -1,3 +1,3 @@
"alias","note","mailboxes"
"ebay@my-domain.com","Used on eBay","destination@my-destionation-domain.com"
"facebook@my-domain.com","Used on Facebook, Instagram.","destination1@my-destionation-domain.com destination2@my-destination-domain.com"
"ebay@my-domain.com","Used on eBay","destination@my-destination-domain.com"
"facebook@my-domain.com","Used on Facebook, Instagram.","destination1@my-destination-domain.com destination2@my-destination-domain.com"
1 alias note mailboxes
2 ebay@my-domain.com Used on eBay destination@my-destionation-domain.com destination@my-destination-domain.com
3 facebook@my-domain.com Used on Facebook, Instagram. destination1@my-destionation-domain.com destination2@my-destination-domain.com destination1@my-destination-domain.com destination2@my-destination-domain.com

217
tests/api/test_import_export.py Normal file
View File

@ -0,0 +1,217 @@
from flask import url_for
from app import alias_utils
from app.extensions import db
from app.models import (
User,
CustomDomain,
Mailbox,
Alias,
AliasMailbox,
BatchImport,
)
from app.import_utils import import_from_csv
from tests.utils import login
def test_export(flask_client):
# Create users
user1 = login(flask_client)
user2 = User.create(
email="x@y.z", password="password", name="Wrong user", activated=True
)
db.session.commit()
# Remove onboarding aliases
for alias in Alias.filter_by(user_id=user1.id).all():
alias_utils.delete_alias(alias, user1)
for alias in Alias.filter_by(user_id=user2.id).all():
alias_utils.delete_alias(alias, user2)
db.session.commit()
# Create domains
CustomDomain.create(
user_id=user1.id, domain="my-destination-domain.com", verified=True
)
CustomDomain.create(
user_id=user2.id, domain="bad-destionation-domain.com", verified=True
)
db.session.commit()
# Create mailboxes
mailbox1 = Mailbox.create(
user_id=user1.id, email="destination@my-destination-domain.com", verified=True
)
mailbox2 = Mailbox.create(
user_id=user1.id, email="destination2@my-destination-domain.com", verified=True
)
badmailbox1 = Mailbox.create(
user_id=user2.id,
email="baddestination@bad-destination-domain.com",
verified=True,
)
db.session.commit()
# Create aliases
Alias.create(
user_id=user1.id,
email="ebay@my-domain.com",
note="Used on eBay",
mailbox_id=mailbox1.id,
)
alias2 = Alias.create(
user_id=user1.id,
email="facebook@my-domain.com",
note="Used on Facebook, Instagram.",
mailbox_id=mailbox1.id,
)
Alias.create(
user_id=user2.id,
email="notmine@my-domain.com",
note="Should not appear",
mailbox_id=badmailbox1.id,
)
db.session.commit()
# Add second mailbox to an alias
AliasMailbox.create(
alias_id=alias2.id,
mailbox_id=mailbox2.id,
)
db.session.commit()
# Export
r = flask_client.get(url_for("api.export_aliases"))
assert r.status_code == 200
assert r.mimetype == "text/csv"
assert (
r.data
== """alias,note,enabled,mailboxes
ebay@my-domain.com,Used on eBay,True,destination@my-destination-domain.com
facebook@my-domain.com,"Used on Facebook, Instagram.",True,destination@my-destination-domain.com destination2@my-destination-domain.com
""".replace(
"\n", "\r\n"
).encode()
)
def test_import_no_mailboxes_no_domains(flask_client):
# Create user
user = login(flask_client)
# Check start state
assert len(Alias.filter_by(user_id=user.id).all()) == 1 # Onboarding alias
alias_data = [
"alias,note",
"ebay@my-domain.com,Used on eBay",
'facebook@my-domain.com,"Used on Facebook, Instagram."',
]
batch_import = BatchImport.create(user_id=user.id, file_id=0)
import_from_csv(batch_import, user, alias_data)
# Should have failed to import anything new because my-domain.com isn't registered
assert len(Alias.filter_by(user_id=user.id).all()) == 1 # +0
def test_import_no_mailboxes(flask_client):
# Create user
user = login(flask_client)
# Check start state
assert len(Alias.filter_by(user_id=user.id).all()) == 1 # Onboarding alias
# Create domain
CustomDomain.create(user_id=user.id, domain="my-domain.com", verified=True)
db.session.commit()
alias_data = [
"alias,note",
"ebay@my-domain.com,Used on eBay",
'facebook@my-domain.com,"Used on Facebook, Instagram."',
]
batch_import = BatchImport.create(user_id=user.id, file_id=0)
import_from_csv(batch_import, user, alias_data)
assert len(Alias.filter_by(user_id=user.id).all()) == 3 # +2
def test_import_no_domains(flask_client):
# Create user
user = login(flask_client)
# Check start state
assert len(Alias.filter_by(user_id=user.id).all()) == 1 # Onboarding alias
alias_data = [
"alias,note,mailboxes",
"ebay@my-domain.com,Used on eBay,destination@my-destination-domain.com",
'facebook@my-domain.com,"Used on Facebook, Instagram.",destination1@my-destination-domain.com destination2@my-destination-domain.com',
]
batch_import = BatchImport.create(user_id=user.id, file_id=0)
import_from_csv(batch_import, user, alias_data)
# Should have failed to import anything new because my-domain.com isn't registered
assert len(Alias.filter_by(user_id=user.id).all()) == 1 # +0
def test_import(flask_client):
# Create user
user = login(flask_client)
# Check start state
assert len(Alias.filter_by(user_id=user.id).all()) == 1 # Onboarding alias
# Create domains
CustomDomain.create(user_id=user.id, domain="my-domain.com", verified=True)
CustomDomain.create(
user_id=user.id, domain="my-destination-domain.com", verified=True
)
db.session.commit()
# Create mailboxes
mailbox1 = Mailbox.create(
user_id=user.id, email="destination@my-destination-domain.com", verified=True
)
mailbox2 = Mailbox.create(
user_id=user.id, email="destination2@my-destination-domain.com", verified=True
)
db.session.commit()
alias_data = [
"alias,note,mailboxes",
"ebay@my-domain.com,Used on eBay,destination@my-destination-domain.com",
'facebook@my-domain.com,"Used on Facebook, Instagram.",destination@my-destination-domain.com destination2@my-destination-domain.com',
]
batch_import = BatchImport.create(user_id=user.id, file_id=0)
import_from_csv(batch_import, user, alias_data)
aliases = Alias.filter_by(user_id=user.id).all()
assert len(aliases) == 3 # +2
# aliases[0] is the onboarding alias, skip it
# eBay alias
assert aliases[1].email == "ebay@my-domain.com"
assert len(aliases[1].mailboxes) == 1
# First one should be primary
assert aliases[1].mailbox_id == mailbox1.id
# Others are sorted
assert aliases[1].mailboxes[0] == mailbox1
# Facebook alias
assert aliases[2].email == "facebook@my-domain.com"
assert len(aliases[2].mailboxes) == 2
# First one should be primary
assert aliases[2].mailbox_id == mailbox1.id
# Others are sorted
assert aliases[2].mailboxes[0] == mailbox2
assert aliases[2].mailboxes[1] == mailbox1