zoo/tests/test_commands.py

# -*- coding: utf-8 -*-

import io
import mock
import pytest
import sys

from zoo.zoo_data.models import Entity, Transaction, Relation
from zoo.zoo_nanterre import utils

from django.core.management import call_command
from django.db.transaction import atomic


def get_output_of_command(command, *args, **kwargs):
    old_stdout = sys.stdout
    output = sys.stdout = io.StringIO()
    call_command(command, *args, **kwargs)
    sys.stdout = old_stdout
    return output.getvalue()


def test_runjobs_command(db):
    call_command('runjobs')

    with mock.patch('zoo.zoo_data.management.commands.runjobs.Job.redo',
                    side_effect=Exception):
        with pytest.raises(Exception):
            call_command('runjobs')


def test_runscript_command(tmpdir):
    script_path = '%s/script.py' % str(tmpdir)
    with open(script_path, 'w') as fd:
        fd.write('# N/A')
    call_command('runscript', script_path)


def test_rsu_cron_command(settings, nanterre_classic_family):
    settings.ZOO_NANTERRE_PASSAGE_A_LA_MAJORITE = True
    call_command('rsu-cron', '-v2')


def test_rsu_dupplicate_command(nanterre_classic_family):
    call_command('rsu-duplicates', '-v2', 'list', '--dedup')
    call_command('rsu-duplicates', '-v2', 'list')

    with pytest.raises(AssertionError, match='Cannot filter a query once a slice has been taken'):
        call_command('rsu-duplicates', 'list', '--count', '2')
    with pytest.raises(TypeError, match="'datetime.datetime' object is not callable"):
        call_command('rsu-duplicates', '-v2', 'list', '--days', '2')

    call_command('rsu-duplicates', '-v2', 'find', '--limit', '0.1')
    call_command('rsu-duplicates', '-v2', 'delete', '--limit', '0.1')


def test_rsu_integrity_check_command(rsu_schema, nanterre_classic_family):
    output = get_output_of_command('rsu-integrity-check')
    assert output == ''

    jean = nanterre_classic_family['jean']
    marie = nanterre_classic_family['marie']
    lilou = nanterre_classic_family['lilou']
    kevin = nanterre_classic_family['kevin']
    with atomic():
        tr = Transaction.get_transaction()
    address2 = Entity.objects.create(
        created=tr,
        schema=rsu_schema[utils.ADRESSE_ENT],
        content={
            'at': '',
            'streetnumber': '169',
            'streetnumberext': '',
            'streetname': 'RUE DU CHATEAU',
            'ext1': '',
            'ext2': '',
            'streetmatriculation': '00169',
            'zipcode': '75014',
            'inseecode': '75014',
            'city': 'PARIS',
            'country': 'FRANCE',
        })
    address3 = Entity.objects.create(
        created=tr,
        schema=rsu_schema[utils.ADRESSE_ENT],
        content={
            'at': '',
            'streetnumber': '170',
            'streetnumberext': 'à côté',
            'streetname': 'RUE DU CHATEAU',
            'ext1': '',
            'ext2': '',
            'streetmatriculation': '00170',
            'zipcode': '75014',
            'inseecode': '75014',
            'city': 'PARIS',
            'country': 'FRANCE',
        })
    Relation.objects.create(
        created=tr,
        schema=rsu_schema[utils.HABITE_REL],
        content={'principale': False},
        left=jean,
        right=address2)
    Relation.objects.create(
        created=tr,
        schema=rsu_schema[utils.HABITE_REL],
        content={'principale': False},
        left=kevin,
        right=address2)
    Relation.objects.create(
        created=tr,
        schema=rsu_schema[utils.HABITE_REL],
        content={'principale': False},
        left=kevin,
        right=address3)
    Relation.objects.create(
        created=tr,
        schema=rsu_schema[utils.UNION_REL],
        content={'status': 'pacs/mariage'},
        left=jean,
        right=marie)
    Relation.objects.create(
        created=tr,
        schema=rsu_schema[utils.RESPONSABILITE_LEGALE_REL],
        content={'status': 'parent'},
        left=lilou,
        right=kevin)
    Relation.objects.filter(
        schema__slug='habite',
        left_id=marie.id
    ).delete()
    Relation.objects.filter(
        schema__slug='habite',
        left_id=lilou.id
    ).delete()
    output = get_output_of_command('rsu-integrity-check')

    assert ("le couple %s / %s est en union plusieurs fois" % (jean.id, marie.id) in output
            or "le couple %s / %s est en union plusieurs fois" % (marie.id, jean.id) in output)
    assert "l'individu %s est en union avec plus d'une personne" % jean.id in output
    assert "l'adulte %s a plus d'une adresse" % jean.id in output
    assert "l'enfant %s a plus de deux parents" % kevin.id in output
    assert "l'enfant %s a plus de deux adresses" % kevin.id in output
    assert "l'adulte %s n'a pas d'adresse" % marie.id in output
    assert "l'enfant %s n'a pas d'adresse" % lilou.id in output


@mock.patch('zoo.zoo_nanterre.utils.psycopg2.connect')
def test_rsu_load_dump_command(mocked_connect, nanterre_classic_family):
    call_command('rsu-load-dump',  '-v3', 'pg_dsn=nanterre_rsu','authentic_fixture_path')