Skip to content

Commit

Permalink
Unittests (#4)
Browse files Browse the repository at this point in the history
* Added Unittests for OAT on lpg

* Adapted Unittests

* Added Unittest for lpg OAT to workflow
  • Loading branch information
TMaesing authored Oct 15, 2023
1 parent ae7a6f9 commit 862c567
Show file tree
Hide file tree
Showing 4 changed files with 143 additions and 14 deletions.
1 change: 1 addition & 0 deletions .github/workflows/python-unittest.yml
Original file line number Diff line number Diff line change
Expand Up @@ -73,3 +73,4 @@ jobs:
python -m unittest up_ac/tests/test_allowed_combinations.py
python -m unittest up_ac/tests/test_enhsp_Irace.py
python -m unittest up_ac/tests/test_fast-downward_Smac.py
python -m unittest up_ac/tests/test_lpg_OAT.py
12 changes: 6 additions & 6 deletions up_ac/tests/test_enhsp_Irace.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,10 +44,10 @@ class TestIraceEnhspOnQuality(unittest.TestCase):
IAC_fb_func = IAC.get_feedback_function(igaci, engine[0],
metric, 'OneshotPlanner')

def test_fb_func(self, IAC_fb_func=IAC_fb_func):
def test_A_fb_func(self, IAC_fb_func=IAC_fb_func):
self.assertIsNotNone(IAC_fb_func, "Operational mode not supported")

def test_enhsp(self, IAC=IAC,
def test_A_enhsp(self, IAC=IAC,
IAC_fb_func=IAC_fb_func, igaci=igaci,
metric=metric, engine=engine, instances=instances):

Expand All @@ -56,7 +56,7 @@ def test_enhsp(self, IAC=IAC,
experiment = {'id.instance': 1, 'configuration': default_config}
IAC_fb_func(experiment, IAC.scenario)

def test_evaluate(self, IAC=IAC, instances=instances, igaci=igaci,
def test_A_evaluate(self, IAC=IAC, instances=instances, igaci=igaci,
engine=engine, metric=metric):

if (IAC.incumbent is None) or (len(instances) == 0):
Expand Down Expand Up @@ -96,10 +96,10 @@ class TestIraceEnhspOnRuntime(unittest.TestCase):
IAC_fb_func = IAC.get_feedback_function(igaci, engine[0],
metric, 'OneshotPlanner')

def test_fb_func(self, IAC_fb_func=IAC_fb_func):
def test_A_fb_func(self, IAC_fb_func=IAC_fb_func):
self.assertIsNotNone(IAC_fb_func, "Operational mode not supported")

def test_enhsp(self, IAC=IAC,
def test_A_enhsp(self, IAC=IAC,
IAC_fb_func=IAC_fb_func, igaci=igaci,
metric=metric, engine=engine, instances=instances):

Expand All @@ -108,7 +108,7 @@ def test_enhsp(self, IAC=IAC,
experiment = {'id.instance': 1, 'configuration': default_config}
IAC_fb_func(experiment, IAC.scenario)

def test_evaluate(self, IAC=IAC, instances=instances, igaci=igaci,
def test_A_evaluate(self, IAC=IAC, instances=instances, igaci=igaci,
engine=engine, metric=metric):

if (IAC.incumbent is None) or (len(instances) == 0):
Expand Down
16 changes: 8 additions & 8 deletions up_ac/tests/test_fast-downward_Smac.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,25 +62,25 @@ class TestSmacFastDownwardOnQuality(unittest.TestCase):
default_config = \
sgaci.engine_param_spaces[engine[0]].get_default_configuration()

def test_fb_func(self, SAC_fb_func=SAC_fb_func, sgaci=sgaci,
def test_A_fb_func(self, SAC_fb_func=SAC_fb_func, sgaci=sgaci,
engine=engine, instances=instances,
default_config=default_config):

self.assertIsNotNone(SAC_fb_func, "Operational mode not supported")
self.assertIsNotNone(SAC_fb_func(default_config, instances[0],
0, reader))

def test_optimize(self, SAC=SAC, SAC_fb_func=SAC_fb_func,
def test_B_optimize(self, SAC=SAC, SAC_fb_func=SAC_fb_func,
default_config=default_config):
incumbent, _ = SAC.optimize(feedback_function=SAC_fb_func)
self.assertIsInstance(incumbent, dict)
self.assertNotEqual(incumbent, default_config)

def test_evaluate(self, metric=metric, engine=engine,
def test_C_evaluate(self, metric=metric, engine=engine,
SAC=SAC, sgaci=sgaci):
perf = SAC.evaluate(metric, engine[0], 'OneshotPlanner',
SAC.incumbent, sgaci, planner_timelimit=5)
self.assertIsNone(perf)
self.assertIsInstance(perf, float)


class TestSmacFastDownwardOnRuntime(unittest.TestCase):
Expand Down Expand Up @@ -127,25 +127,25 @@ class TestSmacFastDownwardOnRuntime(unittest.TestCase):
default_config = \
sgaci.engine_param_spaces[engine[0]].get_default_configuration()

def test_fb_func(self, SAC_fb_func=SAC_fb_func, sgaci=sgaci,
def test_A_fb_func(self, SAC_fb_func=SAC_fb_func, sgaci=sgaci,
engine=engine, instances=instances,
default_config=default_config):

self.assertIsNotNone(SAC_fb_func, "Operational mode not supported")
self.assertIsNotNone(SAC_fb_func(default_config, instances[0],
0, reader))

def test_optimize(self, SAC=SAC, SAC_fb_func=SAC_fb_func,
def test_B_optimize(self, SAC=SAC, SAC_fb_func=SAC_fb_func,
default_config=default_config):
incumbent, _ = SAC.optimize(feedback_function=SAC_fb_func)
self.assertIsInstance(incumbent, dict)
self.assertNotEqual(incumbent, default_config)

def test_evaluate(self, metric=metric, engine=engine,
def test_C_evaluate(self, metric=metric, engine=engine,
SAC=SAC, sgaci=sgaci):
perf = SAC.evaluate(metric, engine[0], 'OneshotPlanner',
SAC.incumbent, sgaci, planner_timelimit=5)
self.assertIsNone(perf)
self.assertIsInstance(perf, float)


up.shortcuts.get_environment().credits_stream = None
Expand Down
128 changes: 128 additions & 0 deletions up_ac/tests/test_lpg_OAT.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,128 @@
import unittest
import unified_planning as up
import sys
import os

# make sure test can be run from anywhere
path = os.getcwd().rsplit('up-ac', 1)[0]
path += 'up-ac/up_ac'
if not os.path.isfile(sys.path[0] + '/configurators.py') and \
'up-ac' in sys.path[0]:
sys.path.insert(0, sys.path[0].rsplit('up-ac', 1)[0] + '/up-ac')

from up_ac.OAT_configurator import OATConfigurator
from up_ac.OAT_interface import OATInterface
from up_ac.utils.download_OAT import get_OAT, delete_OAT, copy_call_engine_OAT

get_OAT()
copy_call_engine_OAT()


class TestOatLpgOnQuality(unittest.TestCase):

get_OAT()
copy_call_engine_OAT()
# pddl instance to test with
instances = [f'{path}/test_problems/visit_precedence/problem.pddl',
f'{path}/test_problems/counters/problem.pddl',
f'{path}/test_problems/depot/problem.pddl',
f'{path}/test_problems/miconic/problem.pddl',
f'{path}/test_problems/matchcellar/problem.pddl']

# test setting
engine = ['lpg']
metric = "quality"

ogaci = OATInterface()
ogaci.read_engine_pcs(engine, f'{path}/engine_pcs')

OAC = OATConfigurator()
OAC.set_training_instance_set(instances)
OAC.set_test_instance_set(instances)

OAC.set_scenario(engine[0],
ogaci.engine_param_spaces[engine[0]], ogaci,
configuration_time=30, n_trials=30,
crash_cost=0, planner_timelimit=15, n_workers=3,
instance_features=None, popSize=5, metric=metric,
evlaLimit=1)

OAC_fb_func = OAC.get_feedback_function(ogaci, engine[0],
metric, 'OneshotPlanner')

def test_A_fb_func(self, OAC_fb_func=OAC_fb_func):
self.assertIsNotNone(OAC_fb_func, "Operational mode not supported")

def test_B_optimize(self, OAC=OAC, OAC_fb_func=OAC_fb_func):
incumbent, _ = OAC.optimize(feedback_function=OAC_fb_func)
self.assertIsInstance(incumbent, dict)

def test_C_evaluate(self, OAC=OAC, metric=metric,
engine=engine, ogaci=ogaci):

perf = OAC.evaluate(metric, engine[0], 'OneshotPlanner',
OAC.incumbent, ogaci)
if OAC.incumbent is None:
self.assertIsNone(perf, "No incumbent and/or no instances should return None")
else:
self.assertIsNotNone(perf, "Should have evaluated")


class TestOatLpgOnRuntime(unittest.TestCase):
# pddl instance to test with
instances = [f'{path}/test_problems/visit_precedence/problem.pddl',
f'{path}/test_problems/counters/problem.pddl',
f'{path}/test_problems/depot/problem.pddl',
f'{path}/test_problems/miconic/problem.pddl',
f'{path}/test_problems/matchcellar/problem.pddl']

# test setting
engine = ['lpg']
metric = "runtime"

ogaci = OATInterface()
ogaci.read_engine_pcs(engine, f'{path}/engine_pcs')

OAC = OATConfigurator()
OAC.set_training_instance_set(instances)
OAC.set_test_instance_set(instances)

OAC.set_scenario(engine[0],
ogaci.engine_param_spaces[engine[0]], ogaci,
configuration_time=30, n_trials=30,
crash_cost=0, planner_timelimit=15, n_workers=3,
instance_features=None, popSize=5, metric=metric,
evlaLimit=1)

OAC_fb_func = OAC.get_feedback_function(ogaci, engine[0],
metric, 'OneshotPlanner')

def test_A_fb_func(self, OAC_fb_func=OAC_fb_func):
self.assertIsNotNone(OAC_fb_func, "Operational mode not supported")

def test_B_optimize(self, OAC=OAC, OAC_fb_func=OAC_fb_func):
incumbent, _ = OAC.optimize(feedback_function=OAC_fb_func)
self.assertIsInstance(incumbent, dict)

def test_C_evaluate(self, OAC=OAC, metric=metric,
engine=engine, ogaci=ogaci):

perf = OAC.evaluate(metric, engine[0], 'OneshotPlanner',
OAC.incumbent, ogaci)
if OAC.incumbent is None:
self.assertIsNone(perf, "No incumbent and/or no instances should return None")
else:
self.assertIsNotNone(perf, "Should have evaluated")

def test_X_delete_OAT(self):
delete_OAT()
path = os.getcwd().rsplit('up_ac', 1)[0]
if path[-1] != "/":
path += "/"
path += 'up_ac'
self.assertFalse(os.path.isdir(f'{path}/OAT/'))


up.shortcuts.get_environment().credits_stream = None
if __name__ == '__main__':
unittest.main()

0 comments on commit 862c567

Please sign in to comment.