Skip to content

Instantly share code, notes, and snippets.

View hpk42's full-sized avatar

holger krekel hpk42

View GitHub Profile
def first_location_match(channel, locations):
import os.path
for x in locations:
if os.path.exists(x):
channel.send(x)
break
else:
channel.send(None)
if __name__ == "__main__":
diff --git a/structlog/threadlocal.py b/structlog/threadlocal.py
index 1b2eeb3..05f40ea 100644
--- a/structlog/threadlocal.py
+++ b/structlog/threadlocal.py
@@ -22,6 +22,34 @@ import uuid
from structlog._config import BoundLoggerLazyProxy
+try:
+ from greenlet import getcurrent as _getident
diff --git a/structlog/threadlocal.py b/structlog/threadlocal.py
index 1b2eeb3..80de726 100644
--- a/structlog/threadlocal.py
+++ b/structlog/threadlocal.py
@@ -17,11 +17,24 @@ Primitives to keep context global but thread local.
"""
import contextlib
-import threading
import uuid
+ def test_is_greenlet_local(self, D):
+ greenlet = pytest.importorskip("greenlet")
+ d = wrap_dict(dict)()
+ d['x'] = 42
+
+ def run():
+ assert 'x' not in d._dict
+ d['x'] = 23
+
import pytest
import time
delay = 0.006
@pytest.mark.parametrize("x", range(100))
def test_func(x):
time.sleep(delay)
import sys
import execnet
gw = execnet.makegateway("popen")
ch = gw.remote_exec("""
while 1:
data = channel.receive()
if data:
# content of testcases/conftest.py
from os.path import abspath, dirname
import os
import pytest
@pytest.fixture(autouse=True)
def cd_to_rootdir():
os.chdir(dirname(dirname(abspath(__file__))))
in case of a whole remote gateway gone:
>>> import execnet
>>> gw=execnet.makegateway()
>>> gw._rinfo()
<RInfo "platform=linux2, executable=/home/hpk/venv/0/bin/python, pid=22601, cwd=/home/hpk/p/devpi, version_info=(2, 7, 3, 'final', 0)">
>>> os.kill(22601, 9)
>>> gw.remote_exec("hello")
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
import pytest
import contextlib
@contextlib.contextmanager
def ctx():
print "setup ctx"
yield 1
print "teardown ctx"
@pytest.fixture
def pytest_load_initial_conftests(parser, args):
ns = parser.parse_known_args(args)
if ns and ns.ds: # orwhatever the option is called
os.environ["DJANGO_SETTINGS"] = ...