#!/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 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 replacePatternInFile(pattern, replace, file_path): try: with fileinput.FileInput(file_path, inplace=True, backup='.bak') as file: for line in file: print(line.replace(pattern, replace), end='') except: return 1 return 0 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 execProcessFromVenv(command: str): 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())) execProcessFromVenv("manage.py flush --noinput") execProcessFromVenv("manage.py makemigrations") execProcessFromVenv("manage.py migrate") execProcessFromVenv("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_is(state:str) -> bool: config = getConfigFromDB() print("config is", config.pyros_state) assert config.pyros_state == state print("Current directory : " + str(os.getcwd())) # Go into src/ os.chdir('..') replacePatternInFile("MAJORDOME_TEST = False", "MAJORDOME_TEST = True", "pyros/settings.py") #os.chdir('majordome') #os.chdir(agent) init_database() sys.path.append('..') print(sys.path) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "src.pyros.settings") django.setup() # 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']) # Launch the 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() print("hello1") print("hello2") time.sleep(10) # # START TESTING Majordome # plcstatus = get_object_or_404(PlcDeviceStatus, id=1) # Now, Majordome should be in state "PASSIVE no PLC" (waiting for PLC) assert_majordome_is("PASSIVE_NO_PLC") # 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" plcstatus.save() time.sleep(5) assert_majordome_is("PASSIVE") # Let's pretend that PLC is now AUTO (via DB) # ==> check that it goes to state "STANDBY" plcstatus.plc_mode = "AUTO" plcstatus.save() time.sleep(10) assert_majordome_is("Standby") # 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() time.sleep(10) assert_majordome_is("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: # 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() time.sleep(10) assert_majordome_is("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) # # Let's set conditions for stopping the Majordome # ==> check that it goes to state config.majordome_state = "RUNNING" config.majordome_state = "STOP" config.save() time.sleep(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") #''' print("hello3") time.sleep(5) print("hello4") #thread_majordome.join() #thread_majordome.killExecutingSequence() os.chdir('..') replacePatternInFile("MAJORDOME_TEST = True", "MAJORDOME_TEST = False", "pyros/settings.py") #os.chdir('majordome')