# -*- coding: utf-8 -*-

from ..syntax import macros, test, test_raises, error, warn, the # noqa: F401

from ..test.fixtures import session, testset

from operator import add

from functools import partial

from collections import deque

from sys import float_info

from queue import Queue

from time import sleep

import threading

import sys

from ..misc import (call, callwith, raisef, tryf, equip_with_traceback,

pack, namelambda, timer,

getattrrec, setattrrec, Popper, CountingIterator, ulp, slurp,

async_raise, callsite_filename, safeissubclass)

from ..fun import withself

from ..env import env

def runtests():

with testset("@call (def as code block)"):

# def as a code block (function overwritten by return value)

@call

def result():

return "hello"

test[result == "hello"]

# use case 1: make temporaries fall out of scope

@call

def x():

a = 2 # many temporaries that help readability...

b = 3 # ...of this calculation, but would just pollute locals...

c = 5 # ...after the block exits

return a * b * c

test[x == 30]

# use case 2: multi-break out of nested loops

@call

def result():

for x in range(10):

for y in range(10):

if x * y == 42:

return (x, y)

... # more code here # pragma: no cover

test[result == (6, 7)]

# can also be used normally

test[the[call(add, 2, 3)] == the[add(2, 3)]]

with testset("@callwith (argument freezer), and pythonic solutions to avoid it"):

# to pass arguments when used as decorator, use @callwith instead

@callwith(3)

def result(x):

return x**2

test[result == 9]

# specialize for given arguments, choose function later

apply23 = callwith(2, 3)

def myadd(a, b):

return a + b

def mymul(a, b):

return a * b

test[apply23(myadd) == 5]

test[apply23(mymul) == 6]

# callwith is not essential; we can do the same pythonically like this:

a = [2, 3]

test[myadd(*a) == 5]

test[mymul(*a) == 6]

# build up the argument list as we go

# - note curry does not help, must use partial; this is because curry

# will happily call "callwith" (and thus terminate the gathering step)

# as soon as it gets at least one argument.

p1 = partial(callwith, 2)

p2 = partial(p1, 3)

p3 = partial(p2, 4)

apply234 = p3() # terminate gathering step by actually calling callwith

def add3(a, b, c):

return a + b + c

def mul3(a, b, c):

return a * b * c

test[apply234(add3) == 9]

test[apply234(mul3) == 24]

# pythonic solution:

a = [2]

a += [3]

a += [4]

test[add3(*a) == 9]

test[mul3(*a) == 24]

# callwith in map, if we want to vary the function instead of the data

m = map(callwith(3), [lambda x: 2 * x, lambda x: x**2, lambda x: x**(1 / 2)])

test[tuple(m) == (6, 9, 3**(1 / 2))]

# pythonic solution - use comprehension notation:

m = (f(3) for f in [lambda x: 2 * x, lambda x: x**2, lambda x: x**(1 / 2)])

test[tuple(m) == (6, 9, 3**(1 / 2))]

# raisef: raise an exception from an expression position

with testset("raisef (raise exception from an expression)"):

raise_instance = lambda: raisef(ValueError("all ok")) # the argument works the same as in `raise ...`

test_raises[ValueError, raise_instance()]

try:

raise_instance()

except ValueError as err:

test[err.__cause__ is None] # like plain `raise ...`, no cause set (default behavior)

# using the `cause` parameter, raisef can also perform a `raise ... from ...`

exc = TypeError("oof")

raise_instance = lambda: raisef(ValueError("all ok"), cause=exc)

test_raises[ValueError, raise_instance()]

try:

raise_instance()

except ValueError as err:

test[err.__cause__ is exc] # cause specified, like `raise ... from ...`

# raisef with old-style parameters (as of v0.14.2, deprecated, will be dropped in v0.15.0)

raise_instance = lambda: raisef(ValueError, "all ok")

test_raises[ValueError, raise_instance()]

# can also raise an exception class (no instance)

test_raises[StopIteration, raisef(StopIteration)]

# tryf: handle an exception in an expression position

with testset("tryf (try/except/finally in an expression)"):

raise_instance = lambda: raisef(ValueError("all ok"))

raise_class = lambda: raisef(ValueError)

test[tryf(lambda: "hello") == "hello"]

test[tryf(lambda: "hello",

elsef=lambda: "there") == "there"]

test[tryf(lambda: raise_instance(),

(ValueError, lambda: "got a ValueError")) == "got a ValueError"]

test[tryf(lambda: raise_instance(),

(ValueError, lambda err: f"got a ValueError: '{err.args[0]}'")) == "got a ValueError: 'all ok'"]

test[tryf(lambda: raise_instance(),

((RuntimeError, ValueError), lambda err: f"got a RuntimeError or ValueError: '{err.args[0]}'")) == "got a RuntimeError or ValueError: 'all ok'"]

test[tryf(lambda: "hello",

(ValueError, lambda: "got a ValueError"),

elsef=lambda: "there") == "there"]

test[tryf(lambda: raisef(ValueError("oof")),

(TypeError, lambda: "got a TypeError"),

((TypeError, ValueError), lambda: "got a TypeError or a ValueError"),

(ValueError, lambda: "got a ValueError")) == "got a TypeError or a ValueError"]

e = env(finally_ran=False)

test[e.finally_ran is False]

test[tryf(lambda: "hello",

elsef=lambda: "there",

finallyf=lambda: e << ("finally_ran", True)) == "there"]

test[e.finally_ran is True]

test[tryf(lambda: raise_class(),

(ValueError, lambda: "ok")) == "ok"]

test[tryf(lambda: raise_class(),

((RuntimeError, ValueError), lambda: "ok")) == "ok"]

test_raises[TypeError, tryf(lambda: "hello",

(str, lambda: "got a string"))] # str is not an exception type

test_raises[TypeError, tryf(lambda: "hello",

((ValueError, str), lambda: "got a string"))] # same, in the tuple case

test_raises[TypeError, tryf(lambda: "hello",

("not a type at all!", lambda: "got a string"))]

with testset("equip_with_traceback"):

e = Exception("just testing")

try:

e = equip_with_traceback(e)

except NotImplementedError:

warn["equip_with_traceback only supported on Python 3.7+, skipping test."]

else:

# Can't do meaningful testing on the result, so just check it's there.

test[e.__traceback__ is not None]

test_raises[TypeError, equip_with_traceback("not an exception")]

with testset("pack"):

myzip = lambda lol: map(pack, *lol)

lol = ((1, 2), (3, 4), (5, 6))

test[tuple(myzip(lol)) == ((1, 3, 5), (2, 4, 6))]

with testset("namelambda"):

square = lambda x: x**2

test[square.__code__.co_name == "<lambda>"]

test[square.__name__ == "<lambda>"]

test[square.__qualname__ == "runtests.<locals>.<lambda>"]

square = namelambda("square")(square)

test[square.__code__.co_name == "square"]

test[square.__name__ == "square"]

test[square.__qualname__ == "runtests.<locals>.square"]

# CAUTION: in case of nested lambdas, the inner doesn't see the outer's new name:

nested = namelambda("outer")(lambda: namelambda("inner")(withself(lambda self: self)))

test[nested.__qualname__ == "runtests.<locals>.outer"]

test[nested().__qualname__ == "runtests.<locals>.<lambda>.<locals>.inner"]

# TODO: Can't raise TypeError; @fploop et al. do-it-now-and-replace-def-with-result

# TODO: decorators need to do this.

test[namelambda("renamed")(42) == 42] # not a function

# simple performance timer as a context manager

with testset("timer"):

with timer() as tictoc:

for _ in range(int(1e6)):

pass

test[tictoc.dt > 0] # elapsed time in seconds (float)

with timer(p=True): # auto-print mode for convenience

for _ in range(int(1e6)):

pass

# access underlying data in an onion of wrappers

with testset("getattrrec, setattrrec (de-onionizers)"):

class Wrapper:

def __init__(self, x):

self.x = x

w = Wrapper(Wrapper(42))

test[type(getattr(w, "x")) == Wrapper]

test[type(getattrrec(w, "x")) == int]

test[getattrrec(w, "x") == 42]

setattrrec(w, "x", 23)

test[type(getattr(w, "x")) == Wrapper]

test[type(getattrrec(w, "x")) == int]

test[getattrrec(w, "x") == 23]

# pop-while iterator

with testset("Popper (pop-while iterator)"):

inp = deque(range(5)) # efficiency: deque can popleft() in O(1) time

out = []

for x in Popper(inp):

out.append(x)

test[inp == deque([])]

test[out == list(range(5))]

inp = deque(range(3))

out = []

for x in Popper(inp):

out.append(x)

if x < 10:

inp.appendleft(x + 10)

test[inp == deque([])]

test[out == [0, 10, 1, 11, 2, 12]]

# works for a list, too, although not efficient (pop(0) takes O(n) time)

inp = list(range(5))

out = []

for x in Popper(inp):

out.append(x)

test[inp == []]

test[out == list(range(5))]

# iterator that counts how many items have been yielded (as a side effect)

with testset("CountingIterator"):

inp = range(5)

it = CountingIterator(inp)

test[it.count == 0]

_ = list(it)

test[it.count == 5]

inp = range(5)

it = CountingIterator(inp)

test[it.count == 0]

for k, _ in enumerate(it, start=1):

test[it.count == k]

test[it.count == 5]

# Unit in the Last Place, float utility

# https://en.wikipedia.org/wiki/Unit_in_the_last_place

with testset("ulp (unit in the last place; float utility)"):

test[ulp(1.0) == float_info.epsilon]

# test also at some base-2 exponent switch points

test[ulp(2.0) == 2 * float_info.epsilon]

test[ulp(0.5) == 0.5 * float_info.epsilon]

with testset("slurp (drain a queue into a list)"):

q = Queue()

for k in range(10):

q.put(k)

test[slurp(q) == list(range(10))]

# async_raise - evil ctypes hack to inject an asynchronous exception into another running thread

if sys.implementation.name != "cpython":

warn["async_raise only supported on CPython, skipping test."] # pragma: no cover

else:

with testset("async_raise (inject KeyboardInterrupt)"):

try:

# Test whether the Python we're running on provides ctypes. At least CPython and PyPy3 do.

# For PyPy3, the general rule is "if it imports, it should work", so let's go along with that.

import ctypes # noqa: F401

out = [] # box, really, but let's not depend on unpythonic.collections in this unrelated unit test module

def test_async_raise_worker():

try:

for j in range(10):

sleep(0.1)

except KeyboardInterrupt: # normally, KeyboardInterrupt is only raised in the main thread

pass

out.append(j)

t = threading.Thread(target=test_async_raise_worker)

t.start()

sleep(0.1) # make sure we're in the while loop

async_raise(t, KeyboardInterrupt)

t.join()

test[out[0] < 9] # terminated early due to the injected KeyboardInterrupt

except NotImplementedError: # pragma: no cover

error["async_raise not supported on this Python interpreter."]

test_raises[TypeError, async_raise(42, KeyboardInterrupt)] # not a thread

t = threading.Thread(target=lambda: None)

t.start()

t.join()

test_raises[ValueError, async_raise(t, KeyboardInterrupt)] # thread no longer running

with testset("callsite_filename"):

test["test_misc.py" in the[callsite_filename()]]

# Like issubclass, but if `cls` is not a class, swallow the `TypeError` and return `False`.

with testset("safeissubclass"):

class MetalBox:

pass

class PlasticBox:

pass

class Safe(MetalBox):

pass

test[safeissubclass(Safe, MetalBox)]

test[not safeissubclass(Safe, PlasticBox)]

test[safeissubclass(Safe, (PlasticBox, MetalBox))]

test[not safeissubclass("definitely not a class", MetalBox)]

if __name__ == '__main__': # pragma: no cover

with session(__file__):

runtests()