Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setup(self):
"""Setup before each method"""
config_data = basic_streamalert_config()
self.fs_patcher = fake_filesystem_unittest.Patcher()
self.fs_patcher.setUp()
self.fs_patcher.fs.create_file(
'/conf/global.json', contents=json.dumps(config_data['global']))
self.fs_patcher.fs.create_file(
'/conf/threat_intel.json', contents=json.dumps(config_data['threat_intel']))
self.fs_patcher.fs.create_file(
'/conf/normalized_types.json', contents=json.dumps(config_data['normalized_types']))
self.fs_patcher.fs.create_file(
'/conf/lambda.json', contents=json.dumps(config_data['lambda']))
self.fs_patcher.fs.create_file(
'/conf/clusters/prod.json', contents=json.dumps(config_data['clusters']['prod']))
# Create the config instance after creating the fake filesystem so that
# CLIConfig uses our mocked config files instead of the real ones.
self.config = CLIConfig()
def setup(self):
"""Setup before each method"""
config_data = basic_streamalert_config()
self.fs_patcher = fake_filesystem_unittest.Patcher()
self.fs_patcher.setUp()
self.fs_patcher.fs.create_file(
'/conf/global.json', contents=json.dumps(config_data['global']))
self.fs_patcher.fs.create_file(
'/conf/threat_intel.json', contents=json.dumps(config_data['threat_intel']))
self.fs_patcher.fs.create_file(
'/conf/normalized_types.json', contents=json.dumps(config_data['normalized_types']))
self.fs_patcher.fs.create_file(
'/conf/lambda.json', contents=json.dumps(config_data['lambda']))
self.fs_patcher.fs.create_file(
'/conf/clusters/prod.json', contents=json.dumps(config_data['clusters']['prod']))
# Create the config instance after creating the fake filesystem so that
# CLIConfig uses our mocked config files instead of the real ones.
self.config = CLIConfig()
def fs(request):
""" Fake filesystem. """
patcher = Patcher()
patcher.setUp()
request.addfinalizer(patcher.tearDown)
return patcher.fs
def setup(self):
"""Setup before each method"""
config_data = basic_streamalert_config()
self.fs_patcher = fake_filesystem_unittest.Patcher()
self.fs_patcher.setUp()
self.fs_patcher.fs.CreateFile('/conf/global.json',
contents=json.dumps(config_data['global']))
self.fs_patcher.fs.CreateFile('/conf/lambda.json',
contents=json.dumps(config_data['lambda']))
self.fs_patcher.fs.CreateFile('/conf/clusters/prod.json',
contents=json.dumps(config_data['clusters']['prod']))
# Create the config instance after creating the fake filesystem so that
# CLIConfig uses our mocked config files instead of the real ones.
self.config = CLIConfig()
def fs(request):
""" Fake filesystem. """
if hasattr(request, 'param'):
# pass optional parameters via @pytest.mark.parametrize
patcher = Patcher(*request.param)
else:
patcher = Patcher()
patcher.setUp()
tokenize._builtin_open = patcher.original_open
yield patcher.fs
patcher.tearDown()
import os
import sys
# Path hacking, for vendored pyfakefs.
sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))
import chardet
import bucketstore
from pyfakefs import fake_filesystem as fake_fs
from pyfakefs.fake_filesystem_unittest import Patcher
from pyfakefs import mox3_stubout, fake_filesystem, fake_pathlib, fake_filesystem_shutil
class S3Patcher(Patcher):
"""docstring for S3Patcher"""
def __init__(self, filesystem):
super(S3Patcher, self).__init__()
self.fs = filesystem
def _refresh(self):
if self._stubs is not None:
self._stubs.smart_unset_all()
self._stubs = mox3_stubout.StubOutForTesting()
for name in self._fake_module_classes:
self._fake_modules[name] = self._fake_module_classes[name](self.fs)
self._fake_modules['path'] = self._fake_modules['os'].path
self.fake_open = fake_filesystem.FakeFileOpen(self.fs)
self._isStale = False