Develop/database fix test

aiaccel/plot.py

@@ -1,6 +1,5 @@
 from aiaccel.argument import Arguments
 from aiaccel.config import Config
-from aiaccel.workspace import Workspace
 from aiaccel.storage.storage import Storage
 from pathlib import Path
 from aiaccel.easy_visualizer import EasyVisualizer

aiaccel/scheduler/job/job_thread.py

@@ -1103,7 +1103,7 @@ def __init__(
         self.storage = Storage(
             self.ws,
             fsmode=self.options['fs'],
-            config_path=self.config_path    
+            config_path=self.config_path
         )
         self.content = self.storage.get_hp_dict(self.trial_id)
         self.result_file_path = self.ws / aiaccel.dict_result / (self.trial_id_str + '.hp')

aiaccel/storage/alive/fs.py

@@ -1,4 +1,3 @@
-from pathlib import PosixPath
 from pathlib import Path
 import aiaccel
 from aiaccel.config import Config

aiaccel/storage/error/fs.py

@@ -1,5 +1,4 @@
 from typing import Union
-from pathlib import PosixPath
 from pathlib import Path
 import aiaccel
 from aiaccel.config import Config

aiaccel/storage/hp/fs.py

@@ -1,4 +1,3 @@
-from pathlib import PosixPath
 from pathlib import Path
 from typing import Any
 from typing import Union
@@ -127,4 +126,5 @@ def all_delete(self) -> None:
         """
         self.update()
         for hp in self.datas.data:
-            hp.remove()
+            if hp is not None:
+                hp.remove()

aiaccel/storage/jobstate/fs.py

@@ -1,4 +1,3 @@
-from pathlib import PosixPath
 from pathlib import Path
 from typing import Union
 import aiaccel

aiaccel/storage/pid/fs.py

@@ -1,5 +1,4 @@
 # import psutil
-from pathlib import PosixPath
 from pathlib import Path
 import aiaccel
 from aiaccel.config import Config

aiaccel/storage/result/fs.py

@@ -1,6 +1,5 @@
 from typing import Any
 from typing import Union
-from pathlib import PosixPath
 from pathlib import Path
 import aiaccel
 from aiaccel.config import Config

aiaccel/storage/serializer/fs.py

@@ -1,5 +1,4 @@
 from typing import Any
-from pathlib import PosixPath
 from pathlib import Path
 from aiaccel.util.snapshot import SnapShot
 from aiaccel.util.retry import retry

aiaccel/storage/timestamp/fs.py

@@ -1,4 +1,3 @@
-from pathlib import PosixPath
 from pathlib import Path
 import aiaccel
 from aiaccel.storage.model.fs import Datalist

aiaccel/storage/trial/fs.py

@@ -1,4 +1,3 @@
-from pathlib import PosixPath
 from pathlib import Path
 from typing import Union
 import aiaccel

aiaccel/util/process.py

@@ -9,6 +9,7 @@
     from aiaccel.master.abci import AbciMaster
     from aiaccel.master.abstract import AbstractMaster
     from aiaccel.master.local import LocalMaster
+import datetime
 
 
 def exec_runner(command: list, silent: bool = True) -> Popen:
@@ -89,6 +90,7 @@ def parse_psaux(outputs: List[bytes]) -> List[dict]:
     return [dict(zip(headers, r)) for r in raw_data]
 
 
+'''
 def ps2joblist() -> List[dict]:
     """Get a ps result and convert to a job list format.
 
@@ -128,6 +130,34 @@ def ps2joblist() -> List[dict]:
         job_list.append(d)
 
     return job_list
+'''
+
+
+def ps2joblist() -> List[dict]:
+    """Get a ps result and convert to a job list format.
+
+    Returns:
+        List[dict]: A job list of ps result.
+
+    Raises:
+        KeyError: Causes when required keys are not contained in a ps result.
+    """
+
+    job_list = []
+
+    for p_info in psutil.process_iter(['pid', 'username', 'status', 'create_time', 'cmdline']):
+        # p_info = proc.as_dict(
+        #    attrs=['pid', 'username', 'status', 'create_time', 'cmdline'])
+        d = {
+            'job-ID': p_info.info['pid'], 'prior': None, 'user': p_info.info['username'],
+            'state': p_info.info['status'], 'queue': None, 'jclass': None,
+            'slots': None, 'ja-task-ID': None, 'name': " ".join(p_info.info['cmdline']),
+            'submit/start at': datetime.datetime.fromtimestamp(
+                p_info.info['create_time']).strftime("%Y-%m-%d %H:%M:%S")
+        }
+        job_list.append(d)
+
+    return job_list
 
 
 def kill_process(pid: int) -> None:

tests/integration/integration_test.py

@@ -23,6 +23,7 @@ async def wait_finish_wrapper(sleep_time, storage, master):
 
 
 def wait_finish(sleep_time, storage, master):
+    time.sleep(sleep_time)
     while (
         storage.alive.check_alive(process_name='master') or
         storage.alive.check_alive(process_name='optimizer') or

tests/resumption/resumption_test.py

@@ -34,17 +34,14 @@ def test_run(self, cd_work, data_dir, work_dir):
         with patch.object(sys, 'argv', commandline_args):
             options = Arguments()
             master = LocalMaster(options)
-
+            master.storage.alive.init_alive()
             run_master(master)
             final_result_at_one_time = get_final_result(work_dir, master)
             print('at one time', final_result_at_one_time)
-            wait_alive(work_dir)
             base_clean_work_dir(data_dir, work_dir)
 
         config_file = data_dir / f'config_{self.search_algorithm}_resumption.json'
 
-        time.sleep(5)
-
         commandline_args = [
             "start.py",
             "--config",
@@ -57,13 +54,14 @@ def test_run(self, cd_work, data_dir, work_dir):
 
         self.create_main()
 
+        time.sleep(5)
+
         with patch.object(sys, 'argv', commandline_args):
             options = Arguments()
             master = LocalMaster(options)
             master.storage.alive.init_alive()
             run_master(master)
             print('resumed steps finished')
-            wait_alive(work_dir)
 
         time.sleep(5)
 
@@ -105,23 +103,6 @@ def run_master(master):
     loop.run_until_complete(gather)
 
 
-def wait_alive(work_dir):
-    alive_files = [
-        work_dir.joinpath(aiaccel.dict_alive, aiaccel.alive_master),
-        work_dir.joinpath(aiaccel.dict_alive, aiaccel.alive_optimizer),
-        work_dir.joinpath(aiaccel.dict_alive, aiaccel.alive_scheduler)
-    ]
-    while True:
-        alive = False
-        for alive_file in alive_files:
-            if fs.check_alive_file(alive_file):
-                alive = True
-                break
-        if not alive:
-            break
-        time.sleep(0.1)
-
-
 def base_clean_work_dir(data_dir, work_dir):
     from tests.conftest import WORK_FILES, WORK_SUB_DIRECTORIES
     import shutil

tests/unit/master_test/test_abstract_master.py

@@ -76,6 +76,10 @@ def test_pre_process(
         if master.optimizer_proc is not None:
             master.optimizer_proc.wait()
 
+        master.worker_o.kill()
+        master.worker_s.kill()
+        master.storage.alive.init_alive()
+
     def test_pre_process_2(
         self,
         cd_work,
@@ -169,6 +173,10 @@ def test_pre_process_3(
         if master.optimizer_proc is not None:
             master.optimizer_proc.wait()
 
+        master.worker_o.kill()
+        master.worker_s.kill()
+        master.storage.alive.init_alive()
+
     def test_post_process(
         self,
         cd_work,
@@ -421,6 +429,10 @@ def test_inner_loop_pre_process(
                 master.pre_process()
                 assert master.inner_loop_pre_process()
 
+        master.worker_o.kill()
+        master.worker_s.kill()
+        master.storage.alive.init_alive()
+
     def test_inner_loop_main_process(
         self,
         cd_work,
@@ -467,6 +479,10 @@ def test_inner_loop_main_process(
         master.get_each_state_count()
         assert not master.inner_loop_main_process()
 
+        master.worker_o.kill()
+        master.worker_s.kill()
+        master.storage.alive.init_alive()
+
     def test_inner_loop_post_process(
         self,
         cd_work,
@@ -499,6 +515,10 @@ def test_inner_loop_post_process(
         master.inner_loop_main_process()
         assert master.inner_loop_post_process()
 
+        master.worker_o.kill()
+        master.worker_s.kill()
+        master.storage.alive.init_alive()
+
     def test_serialize(
         self,
         cd_work,

tests/unit/master_test/verification_test/test_abstract_verification.py

@@ -50,13 +50,20 @@ def test_verify(self, clean_work_dir, setup_hp_finished, work_dir):
                 objective=(i * 1.0)
             )
             verification.storage.trial.set_any_trial_state(trial_id=i, state='finished')
-            for j in range(2):
-                verification.storage.hp.set_any_trial_param(
-                    trial_id=i,
-                    param_name=f'x{j+1}',
-                    param_value=0.0,
-                    param_type='float'
-                )
+            # for j in range(2):
+            # verification.storage.hp.set_any_trial_param(
+            #    trial_id=i,
+            #    param_name=f'x{j+1}',
+            #    param_value=0.0,
+            #    param_type='float'
+            # )
+            verification.storage.hp.set_any_trial_params(
+                trial_id=i,
+                params=[
+                    {'parameter_name': f'x{j+1}', 'value': 0.0, 'type': 'float'}
+                    for j in range(2)
+                ]
+            )
 
         verification.verify()
         file_path = work_dir / aiaccel.dict_verification / f'1.{aiaccel.extension_verification}'

tests/unit/scheduler_test/job/test_job_thread.py

@@ -230,6 +230,7 @@ def test_conditions_result(self, database_remove):
     def test_after_finished(self, database_remove):
         assert self.model.after_finished(self.job) is None
 
+    """
     def test_before_finished(
         self,
         setup_hp_running,
@@ -271,7 +272,73 @@ def test_before_finished(
         print(self.job.trial_id)
         print([d.objective for d in self.job.storage.result.get_all_result()])
         print(self.job.storage.get_best_trial_dict('minimize'))
-
+
+        self.job.next_state = 'finished'
+        self.job.from_file = work_dir.joinpath(aiaccel.dict_hp_running, '001.hp')
+        self.job.to_file = work_dir.joinpath(aiaccel.dict_hp_finished, '001.hp')
+        assert self.model.before_finished(self.job) is None
+    """
+
+    def test_before_finished(
+        self,
+        setup_hp_running,
+        setup_result,
+        work_dir,
+        database_remove
+    ):
+        # setup_hp_running(0)
+        # setup_result(0)
+        # print(self.job.trial_id_str)
+        # print(self.storage.result.get_result_trial_id_list())
+        print(self.job.storage.result.get_all_result())
+        for i in range(10):
+            self.job.storage.result.set_any_trial_objective(trial_id=i, objective=i*1.0)
+            self.job.storage.hp.set_any_trial_params(
+                trial_id=i,
+                params=[
+                    {'parameter_name': f'x{j+1}', 'value': 0.0, 'type': 'float'}
+                    for j in range(10)
+                ]
+            )
+            """
+            for j in range(10):
+                self.job.storage.hp.set_any_trial_param(
+                    trial_id=i,
+                    param_name=f'x{j+1}',
+                    param_value=0.0,
+                    param_type='float'
+                )
+            """
+        assert self.model.before_finished(self.job) is None
+
+        # self.job.storage.trial.all_delete()
+        # self.job.storage.hp.all_delete()
+
+        # setup_hp_running(1)
+        # setup_result(1)
+
+        for i in range(10):
+            self.job.storage.trial.set_any_trial_state(trial_id=i, state='finished')
+            self.job.storage.hp.set_any_trial_params(
+                trial_id=i,
+                params=[
+                    {'parameter_name': f'x{j+1}', 'value': 0.0, 'type': 'float'}
+                    for j in range(10)
+                ]
+            )
+            """
+            for j in range(10):
+                self.job.storage.hp.set_any_trial_param(
+                    trial_id=i,
+                    param_name=f'x{j+1}',
+                    param_value=0.0,
+                    param_type='float'
+                )
+            """
+        print(self.job.trial_id)
+        print([d.objective for d in self.job.storage.result.get_all_result()])
+        print(self.job.storage.get_best_trial_dict('minimize'))
+
         self.job.next_state = 'finished'
         self.job.from_file = work_dir.joinpath(aiaccel.dict_hp_running, '001.hp')
         self.job.to_file = work_dir.joinpath(aiaccel.dict_hp_finished, '001.hp')
@@ -392,7 +459,7 @@ def test_init(
         )
         assert type(job) is Job
 
-    #def test_get_initial_timeout(self):
+    # def test_get_initial_timeout(self):
     #    assert type(self.job.get_initial_timeout()) is datetime.datetime
 
     def test_get_machine(self, database_remove):
@@ -415,6 +482,7 @@ def test_schedule(self, database_remove):
         assert self.job.schedule() is None
 
     def test_run_1(self, database_remove):
+        self.job.scheduler.storage.alive.init_alive()
         self.job.start()
         self.job.join()
 

tests/unit/scheduler_test/test_abstract_scheduler.py

@@ -113,6 +113,7 @@ def test_start_job_thread(
             'process_name': 'scheduler'
         }
         scheduler = AbstractScheduler(options)
+        scheduler.storage.alive.init_alive()
         scheduler.print_dict_state()
         setup_hp_ready(1)
         trial_id = 1

tests/unit/storage/fs/test_fs_alive.py

@@ -1,9 +1,9 @@
 from aiaccel.storage.storage import Storage
 from aiaccel.config import Config
-from base import t_base
-from base import db_path
-from base import ws
-from base import config_path
+from tests.unit.storage.fs.base import t_base
+from tests.unit.storage.fs.base import db_path
+from tests.unit.storage.fs.base import ws
+from tests.unit.storage.fs.base import config_path
 
 
 # init_alive