<commit_before>"""In this test suite, we load the allowlist.json and run through all the tests based
on what expected inputs and return types are provided by the json file.
"""

# stdlib
from itertools import product
import json
import math
import os
from pathlib import Path
import platform
import random
import sys
import time
from typing import Any
from typing import Dict
from typing import Iterable
from typing import List as ListType
from typing import Union

# third party
from packaging import version
import pytest
from pytest import approx
import torch as th

# syft absolute
import syft as sy
from syft.core.pointer.pointer import Pointer
from syft.lib.python import List
from syft.lib.python import String
from syft.lib.python import ValuesIndicesWrapper
from syft.lib.python.primitive_factory import PrimitiveFactory
from syft.lib.python.primitive_factory import isprimitive
from syft.lib.python.primitive_interface import PyPrimitive
from syft.lib.torch import allowlist
from syft.lib.torch.tensor_util import TORCH_STR_DTYPE
from syft.lib.util import full_name_with_qualname

TORCH_VERSION = version.parse(th.__version__.split("+")[0])
py_ver = sys.version_info
PYTHON_VERSION = version.parse(f"{py_ver.major}.{py_ver.minor}")
OS_NAME = platform.system().lower()


# Allow List Test Generation
#
# The purpose of this code is to gather all of the explicitly defined classes,
# attributes, methods and properties that are can be executed remotely via a pointer.
# The vast majority of types and inputs are tensors or collections of numeric values
# so for many cases simply enumerating all possible supported numeric data types
# and using them as both Tensors and inputs covers a lot of the functionality.
# However there are instances where specific methods or properties either do not
# accept input, have specific input requirements or kwargs, or give a variety of output
# types. Additionally some methods exist or behave differently depending on the version
# of the library that we are testing. To support all of this added complexity we start
# with the simplest version which is the "attribute path" to "return type" mapping aka:
#
# allowlist["torch.Tensor.__add__"] = "torch.Tensor"
#
# We then check to see if a explicit configuration is set for the attribute path in
# allowlist_test.json and if so we use the configuration from there.
# To keep the custom definitions DRY we use a concept called "profiles" which are just
# a key lookup in the main JSON tree. Many configuration options also support the
# concept of a key lookup for example you will see lots of "all" which is used to map
# to a whole pre-defined set of supported data types, or inputs. By utilising this
# approach we can prevent the need to maintain a large list of expected exceptions
# which increases the likelihood tests are skipped by accident and makes it very hard
# to track exactly which data types and inputs are supported for a given attribute.
#
# Once we understand what can be tested we construct a product or combination of all
# possible options using the following required variables for each test run:
#
# tensor_type, op_name, self_tensor, _args, is_property, return_type
#
# tensor_type:  this is the data type we convert our test tensor to e.g. uint8
# op_name:      this is the method or property we are testing e.g. __add__
# self_tensor:  this is the tensor we are operating when doing the test
# _args:        this is the input used on the method where "self" represents the
#               self_tensor and None represents no input
# is_property:  this helps understand if the attribute can be called or not
# return_type:  this helps verify that the return type matches the expected return type

# this handles instances where the allow list provides more meta information
def get_return_type(support_dict: Union[str, Dict[str, str]]) -> str:
    if isinstance(support_dict, str):
        return support_dict
    else:
        return support_dict["return_type"]


# this allows us to skip tests based on a few loose matching rules for example:
#
# "skip": [{ "data_types": ["bool"], "inputs": [0, 1] }]
#
# This rule set will match on any test which is for the data type bool and has the
# inputs 0 or 1. Since the primary variation of test combinations relates to their
# tensors, inputs and the data types associated with them
def check_skip(
    combination: ListType,
    skip_rule: Dict[str, Any],
    lib_version: Union[version.Version, version.LegacyVersion],
) -> bool:
    combination_dict = {
        "data_types": combination[0],
        "tensors": combination[2],
        "inputs": combination[3],
    }

    # we will assume the skip rule applies and try to prove it doesn't by finding a
    # condition which we cannot satisfy
    applies = True
    for key in skip_rule.keys():
        # handle the version checks later
        if key.endswith("_version"):
            continue
        if key in combination_dict:
            # first check it is not equal since then the rule would apply
            if combination_dict[key] != skip_rule[key]:
                # second check if the rule is a list of rules in which case we
                # need to check for the presence of our current combination value
                if isinstance(skip_rule[key], Iterable):
                    # we cant use x not in y because bool matches for 0 and 1 in lists
                    presence = any(
                        combination_dict[key] == item
                        and type(combination_dict[key]) == type(item)
                        for item in skip_rule[key]
                    )

                    # if we cant find a matching item of value and type then we dont
                    # match and we wont skip this test, if its True we match so far and
                    # can keep checking more rules
                    applies = presence
                    if not applies:
                        return applies

                else:
                    # we dont match and the rule is not a list of rules so we wont skip
                    # and can break
                    applies = False
                    return applies

    # skip rule versioning example:
    # "skip": [{ "data_types": ["float16"], "lte_version": "1.5.1" }]
    # this says skip float16 if the library is less than or equal to 1.5.1
    # say we are running 1.5.0 that means
    # torch==1.5.0 <= 1.5.1 so this rule applies and we could possibly need to skip

    # if there is a lte_version for the skip rule and our lib is NOT lte, less than or
    # equal to the version then we cant skip because the rule only prohibits lte
    if "lte_version" in skip_rule and (
        not lib_version <= version.parse(skip_rule["lte_version"])
    ):
        applies = False

    # if there is a gte_version for the skip rule and our lib is NOT gte, greater than
    # or equal to the version then  we cant skip because the rule only prohibits gte
    if "gte_version" in skip_rule and (
        not lib_version >= version.parse(skip_rule["gte_version"])
    ):
        applies = False

    # if we were unable to fail the skip rules we should have applies = True and will
    # skip this particular test combination
    return applies


BASIC_OPS = list()
BASIC_OPS_RETURN_TYPE = {}
# here we are loading up the true allowlist which means that we are only testing what
# can be used by the end user
for method, return_type_name_or_dict in allowlist.items():
    if method.startswith("torch.Tensor."):
        return_type = get_return_type(support_dict=return_type_name_or_dict)
        if return_type == "unknown":
            return_type = "torch.Tensor"
        method_name = method.split(".")[-1]
        BASIC_OPS.append(method_name)
        BASIC_OPS_RETURN_TYPE[method_name] = return_type

# load our custom configurations
with open(__file__.replace(".py", ".json"), "r") as f:
    TEST_JSON = json.loads(f.read())

# we need a file to keep all the errors in that makes it easy to debug failures
TARGET_PLATFORM = f"{PYTHON_VERSION}_{TORCH_VERSION}_{OS_NAME}"
ERROR_FILE_PATH = os.path.abspath(
    Path(__file__) / "../../../.." / f"allowlist_test_errors_{TARGET_PLATFORM}.jsonl"
)

# we need a file to keep track of all the methods that are skipped or succeed
SUPPORT_FILE_PATH = os.path.abspath(
    Path(__file__) / "../../../.." / f"allowlist_test_support_{TARGET_PLATFORM}.jsonl"
)


# clear the file before running the tests
if os.path.exists(ERROR_FILE_PATH):
    # this one we can delete since we dont start writing until we are into the tests
    os.unlink(ERROR_FILE_PATH)


# we are running many works in parallel and theres a race condition with deleting this
# file and then writing to it during the collection phase so we are going to just
# spread out the workers and only delete if the file isn't brand new
time.sleep(random.random() * 2)

if os.path.exists(SUPPORT_FILE_PATH):
    # we need to write during gathering so we need to delete this carefully
    try:
        file_stat = os.stat(SUPPORT_FILE_PATH)
        diff = time.time() - file_stat.st_mtime
        if diff > 0.1:
            # only delete on the first run
            for retry in range(5):
                try:
                    os.unlink(SUPPORT_FILE_PATH)
                    break
                except BaseException:
                    time.sleep(1)
    except Exception:
        print(f"Failed while trying to os.stat file {SUPPORT_FILE_PATH}")

<commit_msg>More try catches for Windows CI<commit_after>"""In this test suite, we load the allowlist.json and run through all the tests based
on what expected inputs and return types are provided by the json file.
"""

# stdlib
from itertools import product
import json
import math
import os
from pathlib import Path
import platform
import random
import sys
import time
from typing import Any
from typing import Dict
from typing import Iterable
from typing import List as ListType
from typing import Union

# third party
from packaging import version
import pytest
from pytest import approx
import torch as th

# syft absolute
import syft as sy
from syft.core.pointer.pointer import Pointer
from syft.lib.python import List
from syft.lib.python import String
from syft.lib.python import ValuesIndicesWrapper
from syft.lib.python.primitive_factory import PrimitiveFactory
from syft.lib.python.primitive_factory import isprimitive
from syft.lib.python.primitive_interface import PyPrimitive
from syft.lib.torch import allowlist
from syft.lib.torch.tensor_util import TORCH_STR_DTYPE
from syft.lib.util import full_name_with_qualname

TORCH_VERSION = version.parse(th.__version__.split("+")[0])
py_ver = sys.version_info
PYTHON_VERSION = version.parse(f"{py_ver.major}.{py_ver.minor}")
OS_NAME = platform.system().lower()


# Allow List Test Generation
#
# The purpose of this code is to gather all of the explicitly defined classes,
# attributes, methods and properties that are can be executed remotely via a pointer.
# The vast majority of types and inputs are tensors or collections of numeric values
# so for many cases simply enumerating all possible supported numeric data types
# and using them as both Tensors and inputs covers a lot of the functionality.
# However there are instances where specific methods or properties either do not
# accept input, have specific input requirements or kwargs, or give a variety of output
# types. Additionally some methods exist or behave differently depending on the version
# of the library that we are testing. To support all of this added complexity we start
# with the simplest version which is the "attribute path" to "return type" mapping aka:
#
# allowlist["torch.Tensor.__add__"] = "torch.Tensor"
#
# We then check to see if a explicit configuration is set for the attribute path in
# allowlist_test.json and if so we use the configuration from there.
# To keep the custom definitions DRY we use a concept called "profiles" which are just
# a key lookup in the main JSON tree. Many configuration options also support the
# concept of a key lookup for example you will see lots of "all" which is used to map
# to a whole pre-defined set of supported data types, or inputs. By utilising this
# approach we can prevent the need to maintain a large list of expected exceptions
# which increases the likelihood tests are skipped by accident and makes it very hard
# to track exactly which data types and inputs are supported for a given attribute.
#
# Once we understand what can be tested we construct a product or combination of all
# possible options using the following required variables for each test run:
#
# tensor_type, op_name, self_tensor, _args, is_property, return_type
#
# tensor_type:  this is the data type we convert our test tensor to e.g. uint8
# op_name:      this is the method or property we are testing e.g. __add__
# self_tensor:  this is the tensor we are operating when doing the test
# _args:        this is the input used on the method where "self" represents the
#               self_tensor and None represents no input
# is_property:  this helps understand if the attribute can be called or not
# return_type:  this helps verify that the return type matches the expected return type

# this handles instances where the allow list provides more meta information
def get_return_type(support_dict: Union[str, Dict[str, str]]) -> str:
    if isinstance(support_dict, str):
        return support_dict
    else:
        return support_dict["return_type"]


# this allows us to skip tests based on a few loose matching rules for example:
#
# "skip": [{ "data_types": ["bool"], "inputs": [0, 1] }]
#
# This rule set will match on any test which is for the data type bool and has the
# inputs 0 or 1. Since the primary variation of test combinations relates to their
# tensors, inputs and the data types associated with them
def check_skip(
    combination: ListType,
    skip_rule: Dict[str, Any],
    lib_version: Union[version.Version, version.LegacyVersion],
) -> bool:
    combination_dict = {
        "data_types": combination[0],
        "tensors": combination[2],
        "inputs": combination[3],
    }

    # we will assume the skip rule applies and try to prove it doesn't by finding a
    # condition which we cannot satisfy
    applies = True
    for key in skip_rule.keys():
        # handle the version checks later
        if key.endswith("_version"):
            continue
        if key in combination_dict:
            # first check it is not equal since then the rule would apply
            if combination_dict[key] != skip_rule[key]:
                # second check if the rule is a list of rules in which case we
                # need to check for the presence of our current combination value
                if isinstance(skip_rule[key], Iterable):
                    # we cant use x not in y because bool matches for 0 and 1 in lists
                    presence = any(
                        combination_dict[key] == item
                        and type(combination_dict[key]) == type(item)
                        for item in skip_rule[key]
                    )

                    # if we cant find a matching item of value and type then we dont
                    # match and we wont skip this test, if its True we match so far and
                    # can keep checking more rules
                    applies = presence
                    if not applies:
                        return applies

                else:
                    # we dont match and the rule is not a list of rules so we wont skip
                    # and can break
                    applies = False
                    return applies

    # skip rule versioning example:
    # "skip": [{ "data_types": ["float16"], "lte_version": "1.5.1" }]
    # this says skip float16 if the library is less than or equal to 1.5.1
    # say we are running 1.5.0 that means
    # torch==1.5.0 <= 1.5.1 so this rule applies and we could possibly need to skip

    # if there is a lte_version for the skip rule and our lib is NOT lte, less than or
    # equal to the version then we cant skip because the rule only prohibits lte
    if "lte_version" in skip_rule and (
        not lib_version <= version.parse(skip_rule["lte_version"])
    ):
        applies = False

    # if there is a gte_version for the skip rule and our lib is NOT gte, greater than
    # or equal to the version then  we cant skip because the rule only prohibits gte
    if "gte_version" in skip_rule and (
        not lib_version >= version.parse(skip_rule["gte_version"])
    ):
        applies = False

    # if we were unable to fail the skip rules we should have applies = True and will
    # skip this particular test combination
    return applies


BASIC_OPS = list()
BASIC_OPS_RETURN_TYPE = {}
# here we are loading up the true allowlist which means that we are only testing what
# can be used by the end user
for method, return_type_name_or_dict in allowlist.items():
    if method.startswith("torch.Tensor."):
        return_type = get_return_type(support_dict=return_type_name_or_dict)
        if return_type == "unknown":
            return_type = "torch.Tensor"
        method_name = method.split(".")[-1]
        BASIC_OPS.append(method_name)
        BASIC_OPS_RETURN_TYPE[method_name] = return_type

# load our custom configurations
try:
    with open(__file__.replace(".py", ".json"), "r") as f:
        TEST_JSON = json.loads(f.read())
except Exception as e:
    print(f"Exception {e} triggered")
    raise e

# we need a file to keep all the errors in that makes it easy to debug failures
TARGET_PLATFORM = f"{PYTHON_VERSION}_{TORCH_VERSION}_{OS_NAME}"
ERROR_FILE_PATH = os.path.abspath(
    Path(__file__) / "../../../.." / f"allowlist_test_errors_{TARGET_PLATFORM}.jsonl"
)

# we need a file to keep track of all the methods that are skipped or succeed
SUPPORT_FILE_PATH = os.path.abspath(
    Path(__file__) / "../../../.." / f"allowlist_test_support_{TARGET_PLATFORM}.jsonl"
)


# clear the file before running the tests
if os.path.exists(ERROR_FILE_PATH):
    # this one we can delete since we dont start writing until we are into the tests
    try:
        os.unlink(ERROR_FILE_PATH)
    except Exception as e:
        print(f"Exception {e} triggered")


# we are running many works in parallel and theres a race condition with deleting this
# file and then writing to it during the collection phase so we are going to just
# spread out the workers and only delete if the file isn't brand new
time.sleep(random.random() * 2)

if os.path.exists(SUPPORT_FILE_PATH):
    # we need to write during gathering so we need to delete this carefully
    try:
        file_stat = os.stat(SUPPORT_FILE_PATH)
        diff = time.time() - file_stat.st_mtime
        if diff > 0.1:
            # only delete on the first run
            for retry in range(5):
                try:
                    os.unlink(SUPPORT_FILE_PATH)
                    break
                except BaseException:
                    time.sleep(1)
    except Exception:
        print(f"Failed while trying to os.stat file {SUPPORT_FILE_PATH}")


{'fraction_non_alphanumeric': 0.0631073033120886, 'fraction_numerical': 0.0037613624490648837, 'mean_word_length': 2.9600744724865535, 'pattern_counts': {'":': 20, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2209840', 'n_tokens_mistral': 5117, 'n_tokens_neox': 5019, 'n_words': 2379}