pytorch/caffe2/python/test_util.py
Will Feng cdead5ace1 Enable CircleCI for Linux jobs (#12389)
Summary:
Changes in this PR:
1. Intermediate Docker image is shared from build stage to test stage through ECR, in order to fix the Caffe2 flaky CUDA tests.
2. There are ~7 Caffe2 operator tests that are only flaky in `caffe2_py2_gcc4_8_ubuntu14_04_test` on CPU. Disabling those tests on that config only, which is okay to do because we are still running those tests in other test jobs.

After this PR is merged, CircleCI will be running on master automatically, and will be running on PRs if the author rebased their PR onto the newest master (which we will ask all the authors to do when we switch off Jenkins for Linux).
Pull Request resolved: https://github.com/pytorch/pytorch/pull/12389

Differential Revision: D10224267

Pulled By: yf225

fbshipit-source-id: dd1a90a425c3d13b870d3d328cb301eee2e6e2cd
2018-10-08 17:09:37 -07:00

64 lines
1.7 KiB
Python

## @package test_util
# Module caffe2.python.test_util
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
import numpy as np
from caffe2.python import core, workspace
import unittest
import os
IN_CIRCLECI_FLAKY_ENV = "IN_CIRCLECI" in os.environ and "py2-gcc4.8-ubuntu14.04" in os.environ.get("BUILD_ENVIRONMENT", "")
def rand_array(*dims):
# np.random.rand() returns float instead of 0-dim array, that's why need to
# do some tricks
return np.array(np.random.rand(*dims) - 0.5).astype(np.float32)
def randBlob(name, type, *dims, **kwargs):
offset = kwargs['offset'] if 'offset' in kwargs else 0.0
workspace.FeedBlob(name, np.random.rand(*dims).astype(type) + offset)
def randBlobFloat32(name, *dims, **kwargs):
randBlob(name, np.float32, *dims, **kwargs)
def randBlobsFloat32(names, *dims, **kwargs):
for name in names:
randBlobFloat32(name, *dims, **kwargs)
def numOps(net):
return len(net.Proto().op)
def str_compare(a, b, encoding="utf8"):
if isinstance(a, bytes):
a = a.decode(encoding)
if isinstance(b, bytes):
b = b.decode(encoding)
return a == b
class TestCase(unittest.TestCase):
@classmethod
def setUpClass(cls):
workspace.GlobalInit([
'caffe2',
'--caffe2_log_level=0',
])
# clear the default engines settings to separate out its
# affect from the ops tests
core.SetEnginePref({}, {})
def setUp(self):
self.ws = workspace.C.Workspace()
workspace.ResetWorkspace()
def tearDown(self):
workspace.ResetWorkspace()