1from __future__ import print_function 

2from builtins import str 

3import os 

4import unittest 

5import shutil 

6import unittest 

7import yaml 

8from fundamentals.utKit import utKit 

9from fundamentals import tools 

10from os.path import expanduser 

11import time 

12 

13home = expanduser("~") 

14 

15packageDirectory = utKit("").get_project_root() 

16settingsFile = packageDirectory + "/test_settings.yaml" 

17# settingsFile = home + "/.config/soxspipe.recipes/soxspipe.recipes.yaml" 

18su = tools( 

19 arguments={"settingsFile": settingsFile}, 

20 docString=__doc__, 

21 logLevel="DEBUG", 

22 options_first=False, 

23 projectName=None, 

24 defaultSettingsFile=False 

25) 

26arguments, settings, log, dbConn = su.setup() 

27 

28# SETUP AND TEARDOWN FIXTURE FUNCTIONS FOR THE ENTIRE MODULE 

29moduleDirectory = os.path.dirname(__file__) 

30utKit = utKit(moduleDirectory) 

31log, dbConn, pathToInputDir, pathToOutputDir = utKit.setupModule() 

32utKit.tearDownModule() 

33 

34try: 

35 shutil.rmtree(pathToOutputDir) 

36except: 

37 pass 

38# COPY INPUT TO OUTPUT DIR 

39shutil.copytree(pathToInputDir, pathToOutputDir) 

40 

41# Recursively create missing directories 

42if not os.path.exists(pathToOutputDir): 

43 os.makedirs(pathToOutputDir) 

44 

45# xt-setup-unit-testing-files-and-folders 

46 

47def f(n, anotherKeyword="nothing"): 

48 result = 0 

49 for x in range(10000): 

50 result += n * n * x 

51 return result 

52 

53class test_multiprocess(unittest.TestCase): 

54 

55 def test_multiprocess_function(self): 

56 

57 from fundamentals import fmultiprocess 

58 

59 # DEFINE AN INPUT ARRAY 

60 inputArray = list(range(1000)) 

61 t1 = time.time() 

62 result = fmultiprocess(log=log, function=f, 

63 inputArray=inputArray, anotherKeyword="cheese") 

64 

65 took = time.time() - t1 

66 print("Multiprocessing took: %(took)s" % locals()) 

67 

68 t2 = time.time() 

69 

70 result = [] 

71 for i in inputArray: 

72 result.append(f(i)) 

73 took = time.time() - t2 

74 print("Serial processing took: %(took)s" % locals()) 

75 

76 def test_multiprocess_function_exception(self): 

77 

78 from fundamentals import fmultiprocess 

79 try: 

80 this = fmultiprocess( 

81 log=log, 

82 settings=settings, 

83 fakeKey="break the code" 

84 ) 

85 this.get() 

86 assert False 

87 except Exception as e: 

88 assert True 

89 print(str(e)) 

90 

91 # x-print-testpage-for-pessto-marshall-web-object 

92 

93 # x-class-to-test-named-worker-function