Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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
13home = expanduser("~")
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()
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()
34try:
35 shutil.rmtree(pathToOutputDir)
36except:
37 pass
38# COPY INPUT TO OUTPUT DIR
39shutil.copytree(pathToInputDir, pathToOutputDir)
41# Recursively create missing directories
42if not os.path.exists(pathToOutputDir):
43 os.makedirs(pathToOutputDir)
45# xt-setup-unit-testing-files-and-folders
47def f(n, anotherKeyword="nothing"):
48 result = 0
49 for x in range(10000):
50 result += n * n * x
51 return result
53class test_multiprocess(unittest.TestCase):
55 def test_multiprocess_function(self):
57 from fundamentals import fmultiprocess
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")
65 took = time.time() - t1
66 print("Multiprocessing took: %(took)s" % locals())
68 t2 = time.time()
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())
76 def test_multiprocess_function_exception(self):
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))
91 # x-print-testpage-for-pessto-marshall-web-object
93 # x-class-to-test-named-worker-function