from unittest import TestCase
from prospector.message import Location
class LocationOrderTest(TestCase):
def test_path_order(self):
locs = [
Location('/tmp/path/module3.py', 'module3', 'somefunc', 15, 0),
Location('/tmp/path/module1.py', 'module1', 'somefunc', 10, 0),
Location('/tmp/path/module2.py', 'module2', 'somefunc', 9, 0)
]
paths = [loc.path for loc in locs]
expected = sorted(paths)
self.assertEqual(expected, [loc.path for loc in sorted(locs)])
def test_line_order(self):
locs = [
Location('/tmp/path/module1.py', 'module1', 'somefunc', 15, 0),
Location('/tmp/path/module1.py', 'module1', 'somefunc', 10, 0),
Location('/tmp/path/module1.py', 'module1', 'somefunc', 12, 0)
]
lines = [loc.line for loc in locs]
expected = sorted(lines)
self.assertEqual(expected, [loc.line for loc in sorted(locs)])
def test_sort_between_none_lines(self):
locs = [
Location('/tmp/path/module1.py', 'module1', 'somefunc', 15, 0),
Location('/tmp/path/module1.py', 'module1', 'somefunc', 10, 0),
Location('/tmp/path/module1.py', 'module1', 'somefunc', -1, 0)
]
lines = [(loc.line or -1) for loc in locs]
expected = [None if l == -1 else l for l in sorted(lines)]
self.assertEqual(expected, [loc.line for loc in sorted(locs)])
def test_char_order(self):
locs = [
Location('/tmp/path/module1.py', 'module1', 'somefunc', 10, 7),
Location('/tmp/path/module1.py', 'module1', 'somefunc', 10, 0),
Location('/tmp/path/module1.py', 'module1', 'somefunc', 10, 2)
]
chars = [loc.character for loc in locs]
expected = sorted(chars)
self.assertEqual(expected, [loc.character for loc in sorted(locs)])
def test_sort_between_none_chars(self):
locs = [
Location('/tmp/path/module1.py', 'module1', 'somefunc', 10, -1),
Location('/tmp/path/module1.py', 'module1', 'somefunc', 10, 1),
Location('/tmp/path/module1.py', 'module1', 'somefunc', 10, 2)
]
chars = [(loc.character or -1) for loc in locs]
expected = [None if c == -1 else c for c in sorted(chars)]
self.assertEqual(expected, [loc.character for loc in sorted(locs)])
#!/usr/bin/env python
#
# Copyright (c) 2016 In-Q-Tel, Inc, All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Test module for Onos.py
Created on 28 June 2016
@author: dgrossman
"""
import logging
import logging.config
import pytest
from poseidon.baseClasses.enums_tuples import CRONSPEC
from poseidon.baseClasses.enums_tuples import EVERY
from poseidon.poseidonMain.Scheduler.Scheduler import Scheduler
from poseidon.poseidonMain.Scheduler.Scheduler import scheduler_interface
module_logger = logging.getLogger(__name__)
def test_instantiation():
Scheduler()
def somefunc(jobId, logger):
module_logger.info('someFunc: {0} {1}'.format(jobId, str(logger)))
return True
def test_add():
jobId = 'JOBID'
jobId2 = 'JOBID2'
s = scheduler_interface
s.logger = module_logger
s.logger.setLevel(logging.DEBUG)
b = CRONSPEC(EVERY.minute, None)
ostr = 'cronspec: {0}'.format(str(b))
module_logger.debug(ostr)
s.add_job(jobId, b, somefunc)
s.add_job(jobId2, b, somefunc)
s.schedule.run_all()
assert len(s.schedule.jobs) == 2
s.shutdown()
def test_remove():
jobId = 'JOBID'
jobId2 = 'JOBID2'
s = scheduler_interface
s.logger = module_logger
s.logger.setLevel(logging.DEBUG)
b = CRONSPEC(EVERY.minute, None)
ostr = 'jobs:{0}'.format(len(s.schedule.jobs))
module_logger.info(ostr)
s.add_job(jobId, b, somefunc)
s.add_job(jobId2, b, somefunc)
s.schedule.run_all()
assert len(s.schedule.jobs) == 2
s.del_job(jobId)
assert len(s.schedule.jobs) == 1
s.del_job(jobId2)
assert len(s.schedule.jobs) == 0
s.shutdown()
def test_schedule_once():
jobId = 'JOBID'
jobId2 = 'JOBID2'
s = scheduler_interface
s.logger = module_logger
s.logger.setLevel(logging.DEBUG)
b = CRONSPEC(EVERY.once, '00:00')
c = CRONSPEC(EVERY.once, None)
module_logger.info('jobs:{0}'.format(len(s.schedule.jobs)))
s.add_job(jobId, b, somefunc)
assert len(s.schedule.jobs) == 1
module_logger.info('run')
s.schedule.run_all()
assert len(s.schedule.jobs) == 0
s.add_job(jobId2, c, somefunc)
assert len(s.schedule.jobs) == 1
module_logger.info('run')
s.schedule.run_all()
assert len(s.schedule.jobs) == 0
s.shutdown()
def test_schedule_day():
jobId = 'JOBID'
jobId2 = 'JOBID2'
s = scheduler_interface
s.logger = module_logger
s.logger.setLevel(logging.DEBUG)
b = CRONSPEC(EVERY.day, '00:00')
c = CRONSPEC(EVERY.day, None)
module_logger.info('day jobs:{0}'.format(len(s.schedule.jobs)))
s.add_job(jobId, b, somefunc)
s.add_job(jobId2, c, somefunc)
assert len(s.schedule.jobs) == 2
s.schedule.run_all()
assert len(s.schedule.jobs) == 2
s.del_job(jobId)
assert len(s.schedule.jobs) == 1
s.del_job(jobId2)
assert len(s.schedule.jobs) == 0
s.shutdown()
def test_schedule_hour():
jobId = 'JOBID'
jobId2 = 'JOBID2'
s = scheduler_interface
s.logger = module_logger
s.logger.setLevel(logging.DEBUG)
b = CRONSPEC(EVERY.hour, ':00')
c = CRONSPEC(EVERY.hour, None)
module_logger.info('jobs:{0}'.format(len(s.schedule.jobs)))
s.add_job(jobId, b, somefunc)
s.add_job(jobId2, c, somefunc)
ostr = 'xx' * 20 + '{0}'.format(s.list_jobs())
module_logger.info(ostr)
assert len(s.schedule.jobs) == 2
assert len(s.list_jobs().values()) == 2
s.schedule.run_all()
assert len(s.schedule.jobs) == 2
s.del_job(jobId)
assert len(s.schedule.jobs) == 1
s.del_job(jobId2)
assert len(s.schedule.jobs) == 0
s.shutdown()
def test_schedule_minute():
jobId = 'JOBID'
jobId2 = 'JOBID2'
s = scheduler_interface
s.logger = module_logger
s.logger.setLevel(logging.DEBUG)
b = CRONSPEC(EVERY.minute, None)
c = CRONSPEC(EVERY.minute, 5)
module_logger.info('cronspec:{0}'.format(str(b)))
s.add_job(jobId, b, somefunc)
s.add_job(jobId2, c, somefunc)
s.schedule.run_all()
assert len(s.schedule.jobs) == 2
s.del_job(jobId)
assert len(s.schedule.jobs) == 1
s.del_job(jobId2)
assert len(s.schedule.jobs) == 0
s.shutdown()
'''
Copyright (C) 2018 Swedish Meteorological and Hydrological Institute, SMHI,
This file is part of RAVE.
RAVE is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
RAVE is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with RAVE. If not, see <http://www.gnu.org/licenses/>.
------------------------------------------------------------------------*/
Tests the python version of the pgf registry xml handling
@file
@author Anders Henja (Swedish Meteorological and Hydrological Institute, SMHI)
@date 2018-03-10
'''
import unittest
import string
import os
import rave_pgf_registry
class rave_pgf_registry_test(unittest.TestCase):
TEMPORARY_FILE="rave_pgf_registry_test.xml"
def setUp(self):
if os.path.isfile(self.TEMPORARY_FILE):
os.unlink(self.TEMPORARY_FILE)
def tearDown(self):
if os.path.isfile(self.TEMPORARY_FILE):
os.unlink(self.TEMPORARY_FILE)
def testRead(self):
txt = """<?xml version="1.0" encoding="UTF-8"?>
<generate-registry>
<se.somefunc.1 function="generate" help="help text" module="rave_plugin">
<arguments floats="zra,zrb" ints="a,b,c" strings="str" />
</se.somefunc.1>
<se.somefunc.2 function="generate2" help="help text2" module="rave_plugin2">
<arguments floats="zraz,zrbz" ints="az,bz,cz" strings="strz" />
</se.somefunc.2>
</generate-registry>
"""
self.writeTempFile(txt)
classUnderTest = rave_pgf_registry.PGF_Registry()
classUnderTest.read(self.TEMPORARY_FILE)
self.assertEqual(txt, classUnderTest.tostring())
el = classUnderTest.find("se.somefunc.1")
self.assertTrue(el != None)
self.assertEqual("generate", el.attrib["function"])
self.assertEqual("rave_plugin", el.attrib["module"])
self.assertEqual("help text", el.attrib["help"])
self.assertEqual("zra,zrb", el[0].attrib["floats"])
self.assertEqual("a,b,c", el[0].attrib["ints"])
self.assertEqual("str", el[0].attrib["strings"])
el = classUnderTest.find("se.somefunc.2")
self.assertTrue(el != None)
self.assertEqual("generate2", el.attrib["function"])
self.assertEqual("rave_plugin2", el.attrib["module"])
self.assertEqual("help text2", el.attrib["help"])
self.assertEqual("zraz,zrbz", el[0].attrib["floats"])
self.assertEqual("az,bz,cz", el[0].attrib["ints"])
self.assertEqual("strz", el[0].attrib["strings"])
def testRegister(self):
txt = """<?xml version="1.0" encoding="UTF-8"?>
<generate-registry>
<se.somefunc.1 function="generate" help="help text" module="rave_plugin">
<arguments floats="zra,zrb" ints="a,b,c" strings="str" />
</se.somefunc.1>
<se.somefunc.2 function="generate2" help="help text2" module="rave_plugin2">
<arguments floats="zraz,zrbz" ints="az,bz,cz" strings="strz" />
</se.somefunc.2>
</generate-registry>
"""
expected = """<?xml version="1.0" encoding="UTF-8"?>
<generate-registry>
<se.somefunc.1 function="generate" help="help text" module="rave_plugin">
<arguments floats="zra,zrb" ints="a,b,c" strings="str" />
</se.somefunc.1>
<se.somefunc.2 function="generate2" help="help text2" module="rave_plugin2">
<arguments floats="zraz,zrbz" ints="az,bz,cz" strings="strz" />
</se.somefunc.2>
<se.somefunc.3 function="generate3" help="help text3" module="rave_plugin3">
<arguments floats="f1" ints="i1" strings="s1" />
</se.somefunc.3>
</generate-registry>
"""
self.writeTempFile(txt)
classUnderTest = rave_pgf_registry.PGF_Registry(filename=self.TEMPORARY_FILE)
classUnderTest.register("se.somefunc.3", "rave_plugin3", "generate3", "help text3", "s1", "i1", "f1", "")
with open(self.TEMPORARY_FILE) as fp:
result = fp.read()
self.assertEqual(expected, result)
def writeTempFile(self, txt):
if os.path.isfile(self.TEMPORARY_FILE):
os.unlink(self.TEMPORARY_FILE)
fp = open(self.TEMPORARY_FILE, "w")
fp.write(txt)
fp.close()
Accelerate Your Automation Test Cycles With LambdaTest
Leverage LambdaTest’s cloud-based platform to execute your automation tests in parallel and trim down your test execution time significantly. Your first 100 automation testing minutes are on us.