Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
.addGrid(lasso.regParam, [0.001, 0.005, 0.01]) \
.build()
evaluator = RegressionEvaluator(labelCol="rating", metricName="r2")
cv = CrossValidator(estimator=pipeline, evaluator=evaluator, estimatorParamMaps=paramGrid)
cvModel = cv.fit(data)
test = self.sql.read.format("json").load(trainFilepath)
r2 = evaluator.evaluate(cvModel.transform(test))
print "Test data R^2 score: %g" % r2
if __name__ == "__main__":
unittest.main()
sc.stop()
self.init_debugserver_test()
self.buildDsym()
self.set_inferior_startup_launch()
self.P_and_p_thread_suffix_work()
@llgs_test
@dwarf_test
def test_P_and_p_thread_suffix_work_llgs_dwarf(self):
self.init_llgs_test()
self.buildDwarf()
self.set_inferior_startup_launch()
self.P_and_p_thread_suffix_work()
if __name__ == '__main__':
unittest2.main()
class TestConfProj2(unittest.TestCase):
def test_conf_proj2_with_complexity(self):
main.complexity('tests/conf_proj2')
self.assertTrue(os.path.isfile('tests/conf_proj2/wwwz/index.html'))
self.assertTrue(
os.path.isfile('tests/conf_proj2/wwwz/about/index.html')
)
def tearDown(self):
if os.path.isdir('tests/conf_proj2/wwwz'):
shutil.rmtree('tests/conf_proj2/wwwz')
if __name__ == '__main__':
unittest.main()
expected = Vectors.sparse(n, {hashingTF.indexOf("a"): 1.0,
hashingTF.indexOf("b"): 1.0,
hashingTF.indexOf("c"): 1.0}).toArray()
for i in range(0, n):
self.assertAlmostEqual(output[i], expected[i], 14, "Error at " + str(i) +
": expected " + str(expected[i]) + ", got " + str(output[i]))
if __name__ == "__main__":
from pyspark.mllib.tests import *
if not _have_scipy:
print("NOTE: Skipping SciPy tests as it does not seem to be installed")
if xmlrunner:
unittest.main(testRunner=xmlrunner.XMLTestRunner(output='target/test-reports'))
else:
unittest.main()
if not _have_scipy:
print("NOTE: SciPy tests were skipped as it does not seem to be installed")
sc.stop()
self.assertEqual(N_results, 3)
self.assertEqual(N_messages, 3)
def test_xmldtd_filter(self):
s = results._XMLDTDFilter(BytesIO(b"""Other stuf ab"""))
self.assertEqual(s.read(), b"Other stuf ab")
def test_concatenated_stream(self):
s = results._ConcatenatedStream(BytesIO(b"This is a test "),
BytesIO(b"of the emergency broadcast system."))
self.assertEqual(s.read(3), b"Thi")
self.assertEqual(s.read(20), b's is a test of the e')
self.assertEqual(s.read(), b'mergency broadcast system.')
if __name__ == "__main__":
unittest.main()
def test_handle_enable_host_help(self):
"""Test %s help message""" % handle_enable_host.__name__
self.assert_help(
handle_enable_host,
"""Usage: %s enable-host [options] hostname ...
(Specify the --help global option for a list of other help options)
Options:
-h, --help show this help message and exit
--comment=COMMENT Comment indicating why the host(s) are being enabled
""" % self.progname)
if __name__ == '__main__':
unittest.main()
self.expect("frame info", "We should stop at c()",
substrs = ["a.out`c"])
# Select our parent frame as the current frame.
self.runCmd("frame select 1")
self.expect("frame info", "The immediate caller should be a()",
substrs = ["a.out`a"])
if __name__ == '__main__':
import atexit
lldb.SBDebugger.Initialize()
atexit.register(lambda: lldb.SBDebugger.Terminate())
unittest2.main()
def test_addr_parsing(self):
addr_re = zmq_components.ADDRESS_RE
self.assertEqual(
addr_re.search("ipc:///tmp/foo/bar/baaz-2/foo.0").groups(),
("ipc", "/tmp/foo/bar/baaz-2/foo.0", None))
self.assertEqual(
addr_re.search("tcp://localhost:9000").groups(),
("tcp", "localhost", "9000"))
self.assertEqual(
addr_re.search("tcp://127.0.0.1:9000").groups(),
("tcp", "127.0.0.1", "9000"))
if __name__ == "__main__":
unittest.main()
# Stop the bundle
bnd.stop()
# Ensure the release of the service
self.assertNotIn(bnd, svc_ref.get_using_bundles())
# ------------------------------------------------------------------------------
if __name__ == "__main__":
# Set logging level
import logging
logging.basicConfig(level=logging.DEBUG)
unittest.main()
def main():
unittest.main()