Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_connect_rabbit():
common = Common()
common.connect_rabbit()
def test_get_address_info():
common = Common()
last_update = common.get_address_info('foo', 'bar')
assert last_update == None
def test_setup_env():
common = Common()
common.setup_env()
def __init__(self, files=None, config=None, model=None, model_hash=None, model_path=None):
self.logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
logging.getLogger('pika').setLevel(logging.WARNING)
self.logger = Common().setup_logger(self.logger)
self.common = Common(config=config)
if self.common.use_rabbit:
self.common.connect_rabbit()
self.r = self.common.r
if config:
try:
self.time_const = config['time constant']
self.state_size = config['state size']
self.look_time = config['look time']
self.threshold = config['threshold']
self.rnn_size = config['rnn size']
self.conf_labels = config['conf labels']
self.duration = config['duration']
except Exception as e: # pragma: no cover
self.logger.error(
'Unable to read config properly because: %s', str(e))
def __init__(self, files=None, config=None, model=None, model_hash=None,
model_path=None):
## Initiate logging information on this instance
self.logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
logging.getLogger('pika').setLevel(logging.WARNING)
self.logger = Common().setup_logger(self.logger)
self.common = Common(config=config)
## RabbitMQ acts as a message broker
if self.common.use_rabbit:
self.common.connect_rabbit(host=self.common.rabbit_host,
port=self.common.rabbit_port,
exchange=self.common.rabbit_exchange,
routing_key=self.common.rabbit_routing_key,
queue=self.common.rabbit_queue,
queue_name=self.common.rabbit_queue_name)
## Redis provides a storage capability
if self.common.use_redis:
self.common.connect_redis(host=self.common.redis_host)
if config:
try:
def __init__(self, files=None, config=None, model=None, model_hash=None,
model_path=None):
## Initiate logging information on this instance
self.logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
logging.getLogger('pika').setLevel(logging.WARNING)
self.logger = Common().setup_logger(self.logger)
self.common = Common(config=config)
## RabbitMQ acts as a message broker
if self.common.use_rabbit:
self.common.connect_rabbit(host=self.common.rabbit_host,
port=self.common.rabbit_port,
exchange=self.common.rabbit_exchange,
routing_key=self.common.rabbit_routing_key,
queue=self.common.rabbit_queue,
queue_name=self.common.rabbit_queue_name)
## Redis provides a storage capability
if self.common.use_redis:
self.common.connect_redis(host=self.common.redis_host)
if config:
def __init__(self, files=None, config=None, model=None, model_hash=None, model_path=None):
self.logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
logging.getLogger('pika').setLevel(logging.WARNING)
self.logger = Common().setup_logger(self.logger)
self.common = Common(config=config)
if self.common.use_rabbit:
self.common.connect_rabbit()
self.r = self.common.r
if config:
try:
self.time_const = config['time constant']
self.state_size = config['state size']
self.look_time = config['look time']
self.threshold = config['threshold']
self.rnn_size = config['rnn size']
self.conf_labels = config['conf labels']
self.duration = config['duration']
except Exception as e: # pragma: no cover
self.logger.error(
'Unable to read config properly because: %s', str(e))
def __init__(self):
## Set logging information for instance
self.logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
## Take arguments from command line
self.args = None
self.read_args()
## Take input from configuration file
self.get_config()
self.common = Common(config=self.config)
## Instantiate a logger to to leg messages to aid debugging
self.logger = Common().setup_logger(self.logger)
## Add network traffic files for parsing
self.get_files()
self.model_hash = None
self.model = Model(duration=self.duration, hidden_size=None,
model_type=self.args.algorithm)
def create_base_alg():
return BaseAlgorithm(
files=self.files, config=self.config,
model=self.model, model_hash=self.model_hash,
model_path=self.args.trained_model)
def __init__(self):
## Set logging information for instance
self.logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)
## Take arguments from command line
self.args = None
self.read_args()
## Take input from configuration file
self.get_config()
self.common = Common(config=self.config)
## Instantiate a logger to to leg messages to aid debugging
self.logger = Common().setup_logger(self.logger)
## Add network traffic files for parsing
self.get_files()
self.model_hash = None
self.model = Model(duration=self.duration, hidden_size=None,
model_type=self.args.algorithm)
def create_base_alg():
return BaseAlgorithm(
files=self.files, config=self.config,
model=self.model, model_hash=self.model_hash,
model_path=self.args.trained_model)
## Check whether operation is evaluation, train, or test
## Evaluation returns predictions that are useful for the deployment
## of networkml in an operational environment.