<commit_before>from datetime import datetime
import random
import uuid
from django.test import TestCase
from corehq.apps.sms.handlers.form_session import get_single_open_session_or_close_multiple
from corehq.apps.smsforms.models import SQLXFormsSession, XFormsSession, XFORMS_SESSION_TYPES, XFORMS_SESSION_SMS


class SQLSessionTestCase(TestCase):

    def test_get_by_session_id(self):
        session_id = uuid.uuid4().hex
        sql_session = SQLXFormsSession.objects.create(
            session_id=session_id,
            start_time=datetime.utcnow(),
            modified_time=datetime.utcnow(),
        )
        self.assertEqual(sql_session.pk, SQLXFormsSession.by_session_id(session_id).pk)

    def test_get_by_session_id_not_found(self):
        self.assertEqual(None, SQLXFormsSession.by_session_id(uuid.uuid4().hex))

    def test_sync_from_creation(self):
        properties = _arbitrary_session_properties()
        couch_session = XFormsSession(**properties)
        couch_session.save()
        sql_session = SQLXFormsSession.objects.get(couch_id=couch_session._id)
        for prop, value in properties.items():
            self.assertEqual(getattr(sql_session, prop), value)

        # make sure we didn't do any excess saves
        self.assertTrue(XFormsSession.get_db().get_rev(couch_session._id).startswith('1-'))

    def test_sync_from_update(self):
        properties = _arbitrary_session_properties()
        couch_session = XFormsSession(**properties)
        couch_session.save()
        sql_session = SQLXFormsSession.objects.get(couch_id=couch_session._id)
        for prop, value in properties.items():
            self.assertEqual(getattr(sql_session, prop), value)

        previous_count = SQLXFormsSession.objects.count()
        updated_properties = _arbitrary_session_properties()
        for attr, val in updated_properties.items():
            couch_session[attr] = val
        couch_session.save()

        # make sure nothing new was created
        self.assertEqual(previous_count, SQLXFormsSession.objects.count())
        # check updated props in the sql model
        sql_session = SQLXFormsSession.objects.get(pk=sql_session.pk)
        for prop, value in updated_properties.items():
            self.assertEqual(getattr(sql_session, prop), value)

    def test_reverse_sync(self):
        properties = _arbitrary_session_properties()
        couch_session = XFormsSession(**properties)
        couch_session.save()
        sql_session = SQLXFormsSession.objects.get(couch_id=couch_session._id)
        for prop, value in properties.items():
            self.assertEqual(getattr(sql_session, prop), value)

        # make sure we didn't do any excess saves
        self.assertTrue(XFormsSession.get_db().get_rev(couch_session._id).startswith('1-'))

        updated_properties = _arbitrary_session_properties()
        for prop, value in updated_properties.items():
            setattr(sql_session, prop, value)
        sql_session.save()

        couch_session = XFormsSession.get(couch_session._id)
        for prop, value in updated_properties.items():
            self.assertEqual(getattr(couch_session, prop), value)
        self.assertTrue(couch_session._rev.startswith('2-'))


def _arbitrary_session_properties(**kwargs):
    def arbitrary_string(max_len=32):
        return uuid.uuid4().hex[:max_len]

    def arbitrary_date():
        return datetime(
            random.choice(range(2010, 2015)),
            random.choice(range(1, 13)),
            random.choice(range(1, 28)),
        )

    def arbitrary_bool():
        return random.choice([True, False])

    properties = {
        'connection_id': arbitrary_string(),
        'session_id': arbitrary_string(),
        'form_xmlns': arbitrary_string(),
        'start_time': arbitrary_date(),<commit_msg>add tests for get_single_open_session_or_close_multiple<commit_after>from datetime import datetime
import random
import uuid
from django.test import TestCase
from corehq.apps.sms.handlers.form_session import get_single_open_session_or_close_multiple
from corehq.apps.smsforms.models import SQLXFormsSession, XFormsSession, XFORMS_SESSION_TYPES, XFORMS_SESSION_SMS


class SQLSessionTestCase(TestCase):

    def test_get_by_session_id(self):
        session_id = uuid.uuid4().hex
        sql_session = SQLXFormsSession.objects.create(
            session_id=session_id,
            start_time=datetime.utcnow(),
            modified_time=datetime.utcnow(),
        )
        self.assertEqual(sql_session.pk, SQLXFormsSession.by_session_id(session_id).pk)

    def test_get_by_session_id_not_found(self):
        self.assertEqual(None, SQLXFormsSession.by_session_id(uuid.uuid4().hex))

    def test_sync_from_creation(self):
        properties = _arbitrary_session_properties()
        couch_session = XFormsSession(**properties)
        couch_session.save()
        sql_session = SQLXFormsSession.objects.get(couch_id=couch_session._id)
        for prop, value in properties.items():
            self.assertEqual(getattr(sql_session, prop), value)

        # make sure we didn't do any excess saves
        self.assertTrue(XFormsSession.get_db().get_rev(couch_session._id).startswith('1-'))

    def test_sync_from_update(self):
        properties = _arbitrary_session_properties()
        couch_session = XFormsSession(**properties)
        couch_session.save()
        sql_session = SQLXFormsSession.objects.get(couch_id=couch_session._id)
        for prop, value in properties.items():
            self.assertEqual(getattr(sql_session, prop), value)

        previous_count = SQLXFormsSession.objects.count()
        updated_properties = _arbitrary_session_properties()
        for attr, val in updated_properties.items():
            couch_session[attr] = val
        couch_session.save()

        # make sure nothing new was created
        self.assertEqual(previous_count, SQLXFormsSession.objects.count())
        # check updated props in the sql model
        sql_session = SQLXFormsSession.objects.get(pk=sql_session.pk)
        for prop, value in updated_properties.items():
            self.assertEqual(getattr(sql_session, prop), value)

    def test_reverse_sync(self):
        properties = _arbitrary_session_properties()
        couch_session = XFormsSession(**properties)
        couch_session.save()
        sql_session = SQLXFormsSession.objects.get(couch_id=couch_session._id)
        for prop, value in properties.items():
            self.assertEqual(getattr(sql_session, prop), value)

        # make sure we didn't do any excess saves
        self.assertTrue(XFormsSession.get_db().get_rev(couch_session._id).startswith('1-'))

        updated_properties = _arbitrary_session_properties()
        for prop, value in updated_properties.items():
            setattr(sql_session, prop, value)
        sql_session.save()

        couch_session = XFormsSession.get(couch_session._id)
        for prop, value in updated_properties.items():
            self.assertEqual(getattr(couch_session, prop), value)
        self.assertTrue(couch_session._rev.startswith('2-'))

    def test_get_single_open_session(self):
        properties = _arbitrary_session_properties(
            end_time=None,
            session_type=XFORMS_SESSION_SMS,
        )
        couch_session = XFormsSession(**properties)
        couch_session.save()
        (mult, session) = get_single_open_session_or_close_multiple(
            couch_session.domain, couch_session.connection_id
        )
        self.assertEqual(False, mult)
        self.assertEqual(couch_session._id, session._id)
        [couch_session_back] = XFormsSession.get_all_open_sms_sessions(
            couch_session.domain, couch_session.connection_id
        )
        [sql_session] = SQLXFormsSession.get_all_open_sms_sessions(
            couch_session.domain, couch_session.connection_id
        )
        self.assertEqual(couch_session._id, couch_session_back._id)
        self.assertEqual(couch_session._id, sql_session.couch_id)

    def test_get_single_open_session_close_multiple(self):
        domain = uuid.uuid4().hex
        contact = uuid.uuid4().hex
        for i in range(3):
            properties = _arbitrary_session_properties(
                domain=domain,
                connection_id=contact,
                end_time=None,
                session_type=XFORMS_SESSION_SMS,
            )
            couch_session = XFormsSession(**properties)
            couch_session.save()

        (mult, session) = get_single_open_session_or_close_multiple(domain, contact)
        self.assertEqual(True, mult)
        self.assertEqual(None, session)
        self.assertEqual(0, len(XFormsSession.get_all_open_sms_sessions(domain, contact)))
        self.assertEqual(0, len(SQLXFormsSession.get_all_open_sms_sessions(domain, contact)))


def _arbitrary_session_properties(**kwargs):
    def arbitrary_string(max_len=32):
        return uuid.uuid4().hex[:max_len]

    def arbitrary_date():
        return datetime(
            random.choice(range(2010, 2015)),
            random.choice(range(1, 13)),
            random.choice(range(1, 28)),
        )

    def arbitrary_bool():
        return random.choice([True, False])

    properties = {
        'connection_id': arbitrary_string(),
        'session_id': arbitrary_string(),
        'form_xmlns': arbitrary_string(),
        'start_time': arbitrary_date(),
{'fraction_non_alphanumeric': 0.09261031366294524, 'fraction_numerical': 0.005209994683678894, 'mean_word_length': 3.1860258121940364, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2003793', 'n_tokens_mistral': 2753, 'n_tokens_neox': 2702, 'n_words': 461}