<commit_before>import subprocess, time, os, sys, re

from django.contrib.staticfiles.testing import StaticLiveServerTestCase
from django.test import TestCase
from django.conf import settings

from django.contrib.auth.models import User

from django.core.management.base import OutputWrapper
from django.core.management.color import color_style, no_style

from mqtt.client import BaseClient, MQTTException

from ambulance.models import Ambulance, \
    AmbulanceStatus, AmbulanceCapability

from hospital.models import Hospital, \
    Equipment, HospitalEquipment, EquipmentType

from login.models import AmbulancePermission, HospitalPermission


class MQTTTestCase(StaticLiveServerTestCase):

    @classmethod
    def run_until_success(cls, args, **kwargs):

        # parameters
        MAX_TRIES = kwargs.pop('MAX_TRIES', 10)
        
        # keep trying 
        k = 0
        success = False
        while not success and k < MAX_TRIES:
            if k > 0:
                time.sleep(1)
            k += 1
            retval = subprocess.run(args, **kwargs)
            success = retval.returncode == 0

        if not success:
            raise Exception('Did not succeed!')

    @classmethod
    def run_until_fail(cls, args, **kwargs):

        # parameters
        MAX_TRIES = kwargs.pop('MAX_TRIES', 10)
        
        # keep trying 
        k = 0
        success = True
        while success and k < MAX_TRIES:
            if k > 0:
                time.sleep(1)
            k += 1
            retval = subprocess.run(args, **kwargs)
            success = retval.returncode == 0

        if success:
            raise Exception('Did not fail!')
        
    @classmethod
    def setUpClass(cls):

        try:

            # can get user?
            User.objects.get(username=settings.MQTT['USERNAME'])

        except:

            # Add admin user
            User.objects.create_user(
                username=settings.MQTT['USERNAME'],
                email='<EMAIL>',
                password=settings.MQTT['PASSWORD'],
                is_superuser=True)
        
        # call super to create server
        super().setUpClass()

        # determine server and port
        protocol, host, port = cls.live_server_url.split(':')
        host = host[2:]
        
        print('\n>> Starting django server at {}'.format(cls.live_server_url))
        
        print('>> Stoping mosquitto')
        
        # stop mosquito server
        retval = subprocess.run(["service",
                                 "mosquitto",
                                 "stop"])


        # Wait for shutdown
        cls.run_until_fail(["service",
                            "mosquitto",
                            "status"])
            
        # saving persistence file
        retval = subprocess.run(["mv",
                                 "-f", 
                                 "/var/lib/mosquitto/mosquitto.db",
                                 "/var/lib/mosquitto/mosquitto.db.org"])
        
        # create test configuration file
        with open('/etc/mosquitto/conf.d/test.conf', "w") as outfile:
            
            # change default host and port
            cat = subprocess.Popen(["cat",
                                    "/etc/mosquitto/conf.d/default.conf"],
                                   stdout= subprocess.PIPE)
            sed = subprocess.run(["sed",
                                  "s/8000/{}/".format(port)],
                                 stdin=cat.stdout,
                                 stdout=outfile)
            
        # move current configuration file
        retval = subprocess.run(["mv",
                                 "/etc/mosquitto/conf.d/default.conf",
                                 "/etc/mosquitto/conf.d/default.conf.org"])

        print('>> Start mosquitto with test settings')

        # start mosquito server
        retval = subprocess.run(["service",
                                 "mosquitto",
                                 "start"])

        # Wait for start
        cls.run_until_success(["service",
                                "mosquitto",
                                "status"])
        
        cls.setUpTestData()

    @classmethod
    def tearDownClass(cls):

        # call super to shutdown server
        super().tearDownClass()
        
        print('>> Stopping mosquitto with test settings')
        
        # stop mosquito server
        retval = subprocess.run(["service",
                                 "mosquitto",
                                 "stop"])
        
        # Wait for shutdown
        cls.run_until_fail(["service",
                             "mosquitto",
                             "status"])
        
        # remove test configuration file
        retval = subprocess.run(["mv",
                                 "/etc/mosquitto/conf.d/test.conf",
                                 "/etc/mosquitto/conf.d/test.conf.org"])
        
        # restore current configuration file
        retval = subprocess.run(["mv",
                                 "/etc/mosquitto/conf.d/default.conf.org",
                                 "/etc/mosquitto/conf.d/default.conf"])

        # restore persistence file
        retval = subprocess.run(["mv",
                                 "-f", 
                                 "/var/lib/mosquitto/mosquitto.db.org",
                                 "/var/lib/mosquitto/mosquitto.db"])
        
        print('>> Starting mosquitto')
        
        # start mosquito server
        retval = subprocess.run(["service",
                                 "mosquitto",
                                 "start"])
        
        # Wait for start
        cls.run_until_success(["service",
                                "mosquitto",
                                "status"])
        
    @classmethod
    def setUpTestData(cls):

        # Retrieve admin
        cls.u1 = User.objects.get(username=settings.MQTT['USERNAME'])

        try:
            
            # Add users
            cls.u2 = User.objects.get(username='testuser1')
            cls.u3 = User.objects.get(username='testuser2')

            # Add ambulances
            cls.a1 = Ambulance.objects.get(identifier='BC-179')
            cls.a2 = Ambulance.objects.get(identifier='BC-180')
            cls.a3 = Ambulance.objects.get(identifier='BC-181')

            # Add hospitals
            cls.h1 = Hospital.objects.get(name='Hospital General')
            cls.h2 = Hospital.objects.get(name='Hospital CruzRoja')
            cls.h3 = Hospital.objects.get(name='Hospital Nuevo')<commit_msg>added run_until_*<commit_after>import subprocess, time, os, sys, re

from django.contrib.staticfiles.testing import StaticLiveServerTestCase
from django.test import TestCase
from django.conf import settings

from django.contrib.auth.models import User

from django.core.management.base import OutputWrapper
from django.core.management.color import color_style, no_style

from mqtt.client import BaseClient, MQTTException

from ambulance.models import Ambulance, \
    AmbulanceStatus, AmbulanceCapability

from hospital.models import Hospital, \
    Equipment, HospitalEquipment, EquipmentType

from login.models import AmbulancePermission, HospitalPermission


class MQTTTestCase(StaticLiveServerTestCase):

    @classmethod
    def run_until_success(cls, args, **kwargs):

        # parameters
        MAX_TRIES = kwargs.pop('MAX_TRIES', 10)
        
        # keep trying 
        k = 0
        success = False
        while not success and k < MAX_TRIES:
            if k > 0:
                time.sleep(1)
            k += 1
            retval = subprocess.run(args, **kwargs)
            success = retval.returncode == 0

        if not success:
            raise Exception('Did not succeed!')

    @classmethod
    def run_until_fail(cls, args, **kwargs):

        # parameters
        MAX_TRIES = kwargs.pop('MAX_TRIES', 10)
        
        # keep trying 
        k = 0
        success = True
        while success and k < MAX_TRIES:
            if k > 0:
                time.sleep(1)
            k += 1
            retval = subprocess.run(args, **kwargs)
            success = retval.returncode == 0

        if success:
            raise Exception('Did not fail!')
        
    @classmethod
    def setUpClass(cls):

        try:

            # can get user?
            User.objects.get(username=settings.MQTT['USERNAME'])

        except:

            # Add admin user
            User.objects.create_user(
                username=settings.MQTT['USERNAME'],
                email='<EMAIL>',
                password=settings.MQTT['PASSWORD'],
                is_superuser=True)
        
        # call super to create server
        super().setUpClass()

        # determine server and port
        protocol, host, port = cls.live_server_url.split(':')
        host = host[2:]
        
        print('\n>> Starting django server at {}'.format(cls.live_server_url))
        
        print('>> Stoping mosquitto')
        
        # stop mosquito server
        retval = subprocess.run(["service",
                                 "mosquitto",
                                 "stop"])


        # Wait for shutdown
        cls.run_until_fail(["service",
                            "mosquitto",
                            "status"])
            
        # saving persistence file
        retval = subprocess.run(["mv",
                                 "-f", 
                                 "/var/lib/mosquitto/mosquitto.db",
                                 "/var/lib/mosquitto/mosquitto.db.org"])
        
        # create test configuration file
        with open('/etc/mosquitto/conf.d/test.conf', "w") as outfile:
            
            # change default host and port
            cat = subprocess.Popen(["cat",
                                    "/etc/mosquitto/conf.d/default.conf"],
                                   stdout= subprocess.PIPE)
            sed = subprocess.run(["sed",
                                  "s/8000/{}/".format(port)],
                                 stdin=cat.stdout,
                                 stdout=outfile)
            
        # move current configuration file
        retval = subprocess.run(["mv",
                                 "/etc/mosquitto/conf.d/default.conf",
                                 "/etc/mosquitto/conf.d/default.conf.org"])

        print('>> Start mosquitto with test settings')

        # start mosquito server
        retval = subprocess.run(["service",
                                 "mosquitto",
                                 "start"])

        # Wait for start
        cls.run_until_success(["service",
                                "mosquitto",
                                "status"])


        time.sleep(1)
        
        cls.setUpTestData()

    @classmethod
    def tearDownClass(cls):

        # call super to shutdown server
        super().tearDownClass()
        
        print('>> Stopping mosquitto with test settings')
        
        # stop mosquito server
        retval = subprocess.run(["service",
                                 "mosquitto",
                                 "stop"])
        
        # Wait for shutdown
        cls.run_until_fail(["service",
                             "mosquitto",
                             "status"])
        
        # remove test configuration file
        retval = subprocess.run(["mv",
                                 "/etc/mosquitto/conf.d/test.conf",
                                 "/etc/mosquitto/conf.d/test.conf.org"])
        
        # restore current configuration file
        retval = subprocess.run(["mv",
                                 "/etc/mosquitto/conf.d/default.conf.org",
                                 "/etc/mosquitto/conf.d/default.conf"])

        # restore persistence file
        retval = subprocess.run(["mv",
                                 "-f", 
                                 "/var/lib/mosquitto/mosquitto.db.org",
                                 "/var/lib/mosquitto/mosquitto.db"])
        
        print('>> Starting mosquitto')
        
        # start mosquito server
        retval = subprocess.run(["service",
                                 "mosquitto",
                                 "start"])
        
        # Wait for start
        cls.run_until_success(["service",
                                "mosquitto",
                                "status"])
        
        time.sleep(1)
        
    @classmethod
    def setUpTestData(cls):

        # Retrieve admin
        cls.u1 = User.objects.get(username=settings.MQTT['USERNAME'])

        try:
            
            # Add users
            cls.u2 = User.objects.get(username='testuser1')
            cls.u3 = User.objects.get(username='testuser2')

            # Add ambulances
            cls.a1 = Ambulance.objects.get(identifier='BC-179')
            cls.a2 = Ambulance.objects.get(identifier='BC-180')
            cls.a3 = Ambulance.objects.get(identifier='BC-181')

            # Add hospitals
            cls.h1 = Hospital.objects.get(name='Hospital General')
            cls.h2 = Hospital.objects.get(name='Hospital CruzRoja')
            cls.h3 = Hospital.objects.get(name='Hospital Nuevo')
{'fraction_non_alphanumeric': 0.09958662157083803, 'fraction_numerical': 0.006012777151446824, 'mean_word_length': 1.3032715942530726, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 50, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2601061', 'n_tokens_mistral': 3569, 'n_tokens_neox': 3338, 'n_words': 702}