12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788 |
- import unittest
- import sys
- import os
- import sys
- import time
- import json
- import urllib
- try:
- import urllib2
- except ImportError:
- import urllib as urllib2
- import subprocess
- import signal
- import random
- sys.path.append("interface/HUTN")
- sys.path.append("scripts")
- #from interface.HUTN.hutn_compiler.compiler import main as do_compile
- taskname = "test_task"
- INIT_TIMEOUT = 30
- try:
- import pytest
- slow = pytest.mark.skipif(
- not pytest.config.getoption("--runslow"),
- reason="need --runslow option to run"
- )
- except:
- slow = lambda i:i
- ports = set()
- def getFreePort():
- """Gets a unique new port."""
- while 1:
- port = random.randint(10000, 20000)
- # Check if this port is in the set of ports.
- if port not in ports:
- # We have found a unique port. Add it to the set and return.
- ports.add(port)
- return port
- def execute(scriptname, parameters=[], wait=False):
- if os.name not in ["nt", "posix"]:
- # Stop now, as we would have no clue on how to kill its subtree
- raise Exception("Unknown OS version: " + str(os.name))
- command = [sys.executable, "-u", "scripts/%s.py" % scriptname] + parameters
- if wait:
- return subprocess.call(command, shell=False)
- else:
- return subprocess.Popen(command, shell=False)
- def child_kill(pid):
- subprocess.call(["pkill", "-P", "%i" % pid])
- start = time.time()
- with open(os.devnull, 'w') as null:
- while subprocess.call(["pgrep", "-P", "%i" % pid], stdout=null) > 1:
- time.sleep(0.1)
- if time.time() > start + 4:
- subprocess.call(["pkill", "-9", "-P", "%i" % pid])
- def kill(process):
- if os.name == "nt":
- #TODO this is Windows and I don't care too much about that...
- subprocess.call(["taskkill", "/F", "/T", "/PID", "%i" % process.pid])
- elif os.name == "posix":
- child_kill(process.pid)
- process.send_signal(signal.SIGINT)
- process.wait()
- child_kill(os.getpid())
- def flush_data(address, data):
- if data:
- urllib2.urlopen(urllib2.Request(address, urllib.urlencode({"op": "set_input", "data": json.dumps(data), "taskname": taskname})), timeout=INIT_TIMEOUT).read()
- return []
- def start_mvc():
- port = getFreePort()
- address = "127.0.0.1:%s" % port
- print("Execute run local MV")
- proc = execute("run_local_modelverse", [str(port)], wait=False)
- return proc, address
|