2727import json
2828import os
2929import inspect
30- import shutil
3130import pytest
3231import psutil
3332import shutil
4443from smartsim ._core .config import CONFIG
4544from smartsim .error import SSConfigError
4645from subprocess import run
46+ import sys
4747
4848
4949# Globals, yes, but its a testing file
5050test_path = os .path .dirname (os .path .abspath (__file__ ))
5151test_dir = os .path .join (test_path , "tests" , "test_output" )
5252test_launcher = CONFIG .test_launcher
5353test_device = CONFIG .test_device
54+ test_num_gpus = CONFIG .test_num_gpus
5455test_nic = CONFIG .test_interface
5556test_alloc_specs_path = os .getenv ("SMARTSIM_TEST_ALLOC_SPEC_SHEET_PATH" , None )
5657test_port = CONFIG .test_port
@@ -73,6 +74,7 @@ def print_test_configuration():
7374 global test_nic
7475 global test_alloc_specs_path
7576 global test_port
77+
7678 print ("TEST_SMARTSIM_LOCATION:" , smartsim .__path__ )
7779 print ("TEST_PATH:" , test_path )
7880 print ("TEST_LAUNCHER:" , test_launcher )
@@ -511,7 +513,7 @@ def _test_dir_path(caller_function, caller_fspath):
511513 return dir_path
512514
513515 @staticmethod
514- def get_test_dir (caller_function = None , caller_fspath = None ):
516+ def get_test_dir (caller_function = None , caller_fspath = None , level = 1 ):
515517 """Get path to test output.
516518
517519 This function should be called without arguments from within
@@ -530,7 +532,7 @@ def get_test_dir(caller_function=None, caller_fspath=None):
530532 :rtype: str
531533 """
532534 if not caller_function or not caller_fspath :
533- caller_frame = inspect .stack ()[1 ]
535+ caller_frame = inspect .stack ()[level ]
534536 caller_fspath = caller_frame .filename
535537 caller_function = caller_frame .function
536538
@@ -541,7 +543,7 @@ def get_test_dir(caller_function=None, caller_fspath=None):
541543 return dir_path
542544
543545 @staticmethod
544- def make_test_dir (caller_function = None , caller_fspath = None ):
546+ def make_test_dir (caller_function = None , caller_fspath = None , level = 1 ):
545547 """Create test output directory and return path to it.
546548
547549 This function should be called without arguments from within
@@ -558,7 +560,7 @@ def make_test_dir(caller_function=None, caller_fspath=None):
558560 :rtype: str
559561 """
560562 if not caller_function or not caller_fspath :
561- caller_frame = inspect .stack ()[1 ]
563+ caller_frame = inspect .stack ()[level ]
562564 caller_fspath = caller_frame .filename
563565 caller_function = caller_frame .function
564566
@@ -591,3 +593,38 @@ class MLUtils:
591593 def get_test_device ():
592594 global test_device
593595 return test_device
596+
597+ @staticmethod
598+ def get_test_num_gpus ():
599+ return test_num_gpus
600+
601+ @pytest .fixture
602+ def coloutils ():
603+ return ColoUtils
604+
605+ class ColoUtils :
606+ def setup_test_colo (fileutils , db_type , exp , db_args ):
607+ """Setup things needed for setting up the colo pinning tests"""
608+ # get test setup
609+ test_dir = fileutils .make_test_dir (level = 2 )
610+ sr_test_script = fileutils .get_test_conf_path ("send_data_local_smartredis.py" )
611+
612+ # Create an app with a colo_db which uses 1 db_cpu
613+ colo_settings = exp .create_run_settings (exe = sys .executable , exe_args = sr_test_script )
614+ colo_model = exp .create_model (f"colocated_model" , colo_settings )
615+ colo_model .set_path (test_dir )
616+
617+ if db_type in ['tcp' , "deprecated" ]:
618+ db_args ["port" ] = 6780
619+ db_args ["ifname" ] = "lo"
620+
621+ colocate_fun = {
622+ "tcp" : colo_model .colocate_db_tcp ,
623+ "deprecated" : colo_model .colocate_db ,
624+ "uds" :colo_model .colocate_db_uds
625+ }
626+ colocate_fun [db_type ](** db_args )
627+ # assert model will launch with colocated db
628+ assert colo_model .colocated
629+ # Check to make sure that limit_db_cpus made it into the colo settings
630+ return colo_model
0 commit comments