#!/usr/bin/env python import os import sys import django import subprocess import fileinput import time from django.shortcuts import get_object_or_404 from django.conf import settings as djangosettings import datetime # DONE LATER: #sys.path.append('..') #from common.models import Config, PlcDeviceStatus DEBUG=True agent="majordome" fixture = 'tests/majordome_test' #venv_bin = '../../private/venv_py3_pyros/bin/python' venv_bin = '../private/venv_py3_pyros/bin/python' venv_bin2 = '../../private/venv_py3_pyros/bin/python' def tearDown(self): os.chdir('..') self.replacePatternInFile("MAJORDOME_TEST = True", "MAJORDOME_TEST = False", "pyros/settings.py") os.chdir('majordome') def replacePatternInFileOrError(pattern, replace, file_path): with fileinput.FileInput(file_path, inplace=True, backup='.bak') as file: for line in file: print(line.replace(pattern, replace), end='') # Now, check that replacement has been done or ERROR !!! FOUND=False #with fileinput.FileInput(file_path) as file: with open(file_path) as file: for line in file: #if replace in line: if line.startswith(replace): FOUND = True break #if FOUND: print("pattern "+replace+" found in file "+file_path) if not FOUND: raise(Exception("pattern "+replace+" not found in file "+file_path)) def execProcessAsync(command): #self.printFullTerm(Colors.BLUE, "Executing command [" + command + "]") p = subprocess.Popen(command, shell=True) ''' self.subproc.append((p, command)) self.printFullTerm(Colors.GREEN, "Process launched successfully") self.addExecuted(self.current_command, command) ''' return p def execProcessFromVenvSync(command: str): print(f"Execute command from venv as a process: {command}") command = venv_bin + ' ' + command args = command.split() process = subprocess.Popen(args) process.wait() if process.returncode == 0: print("Process executed successfully") else: print("Process execution failed") return process.returncode def execProcessFromVenvAsync(command: str): command = venv_bin2 + ' ' + command args = command.split() #self.printFullTerm(Colors.BLUE, "Executing command from venv [" + str(' '.join(args[1:])) + "]") p = subprocess.Popen(args) #self.subproc.append((p, ' '.join(args[1:]))) #self.printFullTerm(Colors.GREEN, "Process launched successfully") #self.addExecuted(self.current_command, str(' '.join(args[1:]))) return p def init_database(): # Go into src/ #os.chdir('..') print("Current directory : " + str(os.getcwd())) # If executed the very first time, it raises errors, so executes it later #execProcessFromVenvSync("manage.py flush --noinput") execProcessFromVenvSync("manage.py makemigrations") execProcessFromVenvSync("manage.py migrate") # Executed here, it is ok execProcessFromVenvSync("manage.py flush --noinput") # Do we really need now to make migrations and to migrate again (a second time) ? (not sure) execProcessFromVenvSync("manage.py makemigrations") execProcessFromVenvSync("manage.py migrate") # Load fixture: execProcessFromVenvSync("manage.py loaddata misc" + os.sep + "fixtures" + os.sep + fixture) def getConfigFromDB(): return get_object_or_404(Config, id=1) def changeDirectory(path): #if DEBUG: print("Moving to : " + path) os.chdir(path) if DEBUG: print("NEW Current directory : " + str(os.getcwd())) def assert_majordome_has_now_state(state:str) -> bool: config = getConfigFromDB() print() #print("Majordome state should now be", config.pyros_state) print("=> Majordome state should now be", state) assert config.pyros_state == state print("=> CHECKED") print() def print_step_message(step_num:int, message:str): print() print('--------------') print(f'{step_num}) {message}') print('--------------') def go_sleeping(nbsec:int): print(f"GO TO SLEEP {nbsec} seconds...") time.sleep(nbsec) print("...STOP SLEEPING") i=0 print("Current directory : " + str(os.getcwd())) i+=1; print_step_message(i, 'Change pyros/settings just for this test, so that it uses test database') # Go into src/ os.chdir('..') replacePatternInFileOrError("MAJORDOME_TEST = False", "MAJORDOME_TEST = True", "pyros/settings.py") #os.chdir('majordome') i+=1; print_step_message(i, 'Init test database') init_database() i+=1; print_step_message(i, 'Django setup') sys.path.append('..') #print(sys.path) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "src.pyros.settings") django.setup() assert djangosettings.DATABASES['default']['NAME'] == 'pyros_test' i+=1; print_step_message(i, 'Get config') # Needs sys.path.append('..') from common.models import Config, PlcDeviceStatus config = get_object_or_404(Config, id=1) print("config id is", config.id) print("config latitude is", config.latitude) print("config global mode is", config.global_mode) print("config row_data_save_frequency is", config.row_data_save_frequency) #with print_lock: print("global mode is", getConfigFromDB().global_mode) #sys.stdout.write("global mode") #sys.stdout.flush() print("global mode is", getConfigFromDB().global_mode) print("DB1 used is:", djangosettings.DATABASES['default']['NAME']) go_sleeping(2) # ... ''' # Launch the Majordome agent ''' i+=1; print_step_message(i, 'Launch Majordome agent') os.chdir(agent) #p = execProcessFromVenvAsync(venv_bin + ' start_agent_'+agent+'_from_test.py') #p = execProcessFromVenvAsync(venv_bin + ' start_agent_'+agent+'.py') majordome = execProcessFromVenvAsync("start_agent_"+agent+'.py') #p = self.execProcessFromVenvAsync('./start_agent_'+agent+'.py') #p.wait() go_sleeping(10) # ... ''' # START TESTING Majordome ''' i+=1; print_step_message(i, 'Start testing Majordome states') plcstatus = get_object_or_404(PlcDeviceStatus, id=1) # Now, Majordome should be in state "PASSIVE no PLC" (waiting for PLC) assert_majordome_has_now_state("PASSIVE_NO_PLC") print("\n- I now simulate that PLC is OK:") # Let's pretend that PLC is now OK (via DB) # ==> check that it goes to state "PASSIVE" (waiting for PLC "AUTO") #plcstatus.created = "2018-07-24 11:49:49" # ex: '2018-07-27T10:12:38.875112' plcstatus.created = datetime.datetime.now().isoformat() plcstatus.save() go_sleeping(5) # ... assert_majordome_has_now_state("PASSIVE") print("\n- I now simulate that PLC is AUTO:") # Let's pretend that PLC is now AUTO (via DB) # ==> check that it goes to state "STANDBY" plcstatus.plc_mode = "AUTO" plcstatus.save() go_sleeping(15) # ... assert_majordome_has_now_state("Standby") print("\n- I now simulate that Majordome is going REMOTE mode:") # Let's pretend that we are going REMOTE mode # ==> check that it goes to state "REMOTE" config.global_mode = False config.lock = False config.save() go_sleeping(10) # ... assert_majordome_has_now_state("Remote") # Let's pretend that PLC is now UNSAFE # ==> check that shutter is closing #TODO: #plc_status.is_safe # Let's pretend that PLC is now SAFE again # ==> check that shutter is opening #TODO: print("\n- I now simulate that Majordome is going SCHEDULER mode:") # Let's pretend that we are going SCHEDULER mode # ==> check that it goes to state "STANDBY" config.global_mode = True config.lock = False config.save() go_sleeping(10) # ... assert_majordome_has_now_state("Standby") # Let's set conditions for going to SCHEDULER mode # ==> check that it goes to state "SCHEDULER" #TODO: # self.is_night() and self.plc_is_safe() and self.config.ack and not self.config.lock # Let's set conditions for leaving the SCHEDULER mode # ==> check that it goes to state "STANDBY" #TODO: ''' # KILL agent Majordome (using DB config table) ''' i+=1; print_step_message(i, 'Stop (kill) Majordome agent process') # Let's set conditions for stopping the Majordome # ==> check that it goes to state config.majordome_state = "RUNNING" config.majordome_state = "STOP" config.save() go_sleeping(5) # ... config = getConfigFromDB() print("config is", config.majordome_state) assert config.majordome_state == "RUNNING" ##majordome.kill() #self.execProcessAsync("ps aux | grep \"start_agent_majordome.py\" | awk '{ print $2 }' | xargs kill") #self.execProcessAsync("ps aux | grep start_agent_majordome.py | awk '{ print $2 }' | xargs kill") #''' go_sleeping(5) # ... #thread_majordome.join() #thread_majordome.killExecutingSequence() i+=1; print_step_message(i, 'Restore pyros/settings as before') os.chdir('..') replacePatternInFileOrError("MAJORDOME_TEST = True", "MAJORDOME_TEST = False", "pyros/settings.py") #os.chdir('majordome')