Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from datetime import timedelta
from time import sleep
import unittest
import pytest
from rx.scheduler.mainloop import WxScheduler
from rx.internal.basic import default_now
wx = pytest.importorskip('wx')
class AppExit(wx.Timer):
def __init__(self, app) -> None:
super().__init__()
self.app = app
def Notify(self):
self.app.ExitMainLoop()
class TestWxScheduler(unittest.TestCase):
def test_wx_schedule_now(self):
scheduler = WxScheduler(wx)
def test_xdist_normal_count(self, many_tests_files, testdir, monkeypatch):
pytest.importorskip("xdist")
monkeypatch.delenv("PYTEST_DISABLE_PLUGIN_AUTOLOAD", raising=False)
testdir.makeini(
"""
[pytest]
console_output_style = count
"""
)
output = testdir.runpytest("-n2")
output.stdout.re_match_lines([r"\.{20} \s+ \[20/20\]"])
def engine(self, request):
module_name, dsn = request.param
# check module is available
pytest.importorskip(module_name)
engine = create_engine(dsn, poolclass=StaticPool)
try:
engine.connect()
except OperationalError:
pytest.skip('could not connect to database {}'.format(dsn))
return engine
def test_consolidate_singles(self):
pytest.importorskip('scipy')
cp_true = np.array([1,1,1,0,0])
self.g.append_cp(consolidate_singles=True)
cp_test = g.v.cp.values
assert_allclose(cp_true, cp_test)
def test_nc_load(dtype):
pytest.importorskip('netCDF4')
truth = xarray.open_dataset(R/'r2all.nc', group=dtype)
obs = gr.load(R/f'demo.10{dtype[0].lower()}')
assert obs.equals(truth)
def sspi(self):
# these testa are run on localhost and require a valid Windows user
# set under CREDSSP_USERNAME and CREDSSP_PASSWORD
pytest.importorskip("sspi")
username = os.environ.get('CREDSSP_USERNAME', None)
password = os.environ.get('CREDSSP_PASSWORD', None)
if username and password:
server_context = sspi.ServerAuth(
pkg_name="Negotiate",
)
return server_context, "127.0.0.0", username, password
else:
pytest.skip("CREDSSP_USERNAME, CREDSSP_PASSWORD, CREDSSP_SERVER "
"environment variables were not set, integration tests"
def test_sqlalchemy_engine_ref():
global sqla_engine
sqlalchemy = pytest.importorskip('apscheduler.jobstores.sqlalchemy')
sqla_engine = sqlalchemy.create_engine('sqlite:///')
try:
sqlalchemy.SQLAlchemyJobStore(engine='%s:sqla_engine' % __name__)
finally:
sqla_engine.dispose()
del sqla_engine
# Copyright: 2011 MoinMoin:RonnyPfannschmidt
# Copyright: 2011 MoinMoin:ThomasWaldmann
# License: GNU GPL v2 (or any later version), see LICENSE.txt for details.
"""
MoinMoin - kyoto tycoon store tests
"""
from __future__ import absolute_import, division
import pytest
pytest.importorskip('MoinMoin.storage.stores.kt')
from MoinMoin._tests import check_connection
try:
check_connection(1978)
except Exception as err:
pytestmark = pytest.mark.skip(str(err))
from ..kt import BytesStore, FileStore
@pytest.mark.multi(Store=[BytesStore, FileStore])
def test_create(Store):
store = Store()
store.create()
def test_load_hires():
pytest.importorskip('netCDF4')
img = gq.load(R / 'goes13.2017.233.080017.BAND_02.nc')
assert img.shape == (1100, 2500)
def test_sum(ctx_factory):
from pytest import importorskip
importorskip("mako")
context = ctx_factory()
queue = cl.CommandQueue(context)
n = 200000
for dtype in [np.float32, np.complex64]:
a_gpu = general_clrand(queue, (n,), dtype)
a = a_gpu.get()
for slc in [
slice(None),
slice(1000, 3000),
slice(1000, -3000),
slice(1000, None),
slice(1000, None, 3),