text:
from unittest import mock

from django.contrib.postgres.indexes import (
    BloomIndex, BrinIndex, BTreeIndex, GinIndex, GistIndex, HashIndex, OpClass,
    SpGistIndex,
)
from django.db import NotSupportedError, connection
from django.db.models import CharField, F, Index, Q
from django.db.models.functions import Cast, Collate, Length, Lower
from django.test import skipUnlessDBFeature
from django.test.utils import modify_settings, register_lookup

from . import PostgreSQLSimpleTestCase, PostgreSQLTestCase
from .fields import SearchVector, SearchVectorField
from .models import CharFieldModel, IntegerArrayModel, Scene, TextFieldModel

class IndexTestMixin:

    def test_name_auto_generation(self):
        index = self.index_class(fields=['field'])
        index.set_name_with_model(CharFieldModel)
        self.assertRegex(index.name, r'postgres_te_field_[0-9a-f]{6}_%s' % self.index_class.suffix)

    def test_deconstruction_no_customization(self):
        index = self.index_class(fields=['title'], name='test_title_%s' % self.index_class.suffix)
        path, args, kwargs = index.deconstruct()
        self.assertEqual(path, 'django.contrib.postgres.indexes.%s' % self.index_class.__name__)
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {'fields': ['title'], 'name': 'test_title_%s' % self.index_class.suffix})

    def test_deconstruction_with_expressions_no_customization(self):
        name = f'test_title_{self.index_class.suffix}'
        index = self.index_class(Lower('title'), name=name)
        path, args, kwargs = index.deconstruct()
        self.assertEqual(
            path,
            f'django.contrib.postgres.indexes.{self.index_class.__name__}',
        )
        self.assertEqual(args, (Lower('title'),))
        self.assertEqual(kwargs, {'name': name})

class BloomIndexTests(IndexTestMixin, PostgreSQLSimpleTestCase):
    index_class = BloomIndex

    def test_suffix(self):
        self.assertEqual(BloomIndex.suffix, 'bloom')

    def test_deconstruction(self):
        index = BloomIndex(fields=['title'], name='test_bloom', length=80, columns=[4])
        path, args, kwargs = index.deconstruct()
        self.assertEqual(path, 'django.contrib.postgres.indexes.BloomIndex')
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {
            'fields': ['title'],
            'name': 'test_bloom',
            'length': 80,
            'columns': [4],
        })

    def test_invalid_fields(self):
        msg = 'Bloom indexes support a maximum of 32 fields.'
        with self.assertRaisesMessage(ValueError, msg):
            BloomIndex(fields=['title'] * 33, name='test_bloom')

    def test_invalid_columns(self):
        msg = 'BloomIndex.columns must be a list or tuple.'
        with self.assertRaisesMessage(ValueError, msg):
            BloomIndex(fields=['title'], name='test_bloom', columns='x')
        msg = 'BloomIndex.columns cannot have more values than fields.'
        with self.assertRaisesMessage(ValueError, msg):
            BloomIndex(fields=['title'], name='test_bloom', columns=[4, 3])

    def test_invalid_columns_value(self):
        msg = 'BloomIndex.columns must contain integers from 1 to 4095.'
        for length in (0, 4096):
            with self.subTest(length), self.assertRaisesMessage(ValueError, msg):
                BloomIndex(fields=['title'], name='test_bloom', columns=[length])

    def test_invalid_length(self):
        msg = 'BloomIndex.length must be None or an integer from 1 to 4096.'
        for length in (0, 4097):
            with self.subTest(length), self.assertRaisesMessage(ValueError, msg):
                BloomIndex(fields=['title'], name='test_bloom', length=length)

class BrinIndexTests(IndexTestMixin, PostgreSQLSimpleTestCase):
    index_class = BrinIndex

    def test_suffix(self):
        self.assertEqual(BrinIndex.suffix, 'brin')

    def test_deconstruction(self):
        index = BrinIndex(fields=['title'], name='test_title_brin', autosummarize=True, pages_per_range=16)
        path, args, kwargs = index.deconstruct()
        self.assertEqual(path, 'django.contrib.postgres.indexes.BrinIndex')
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {
            'fields': ['title'],
            'name': 'test_title_brin',
            'autosummarize': True,
            'pages_per_range': 16,
        })

    def test_invalid_pages_per_range(self):
        with self.assertRaisesMessage(ValueError, 'pages_per_range must be None or a positive integer'):
            BrinIndex(fields=['title'], name='test_title_brin', pages_per_range=0)

class BTreeIndexTests(IndexTestMixin, PostgreSQLSimpleTestCase):
    index_class = BTreeIndex

    def test_suffix(self):
        self.assertEqual(BTreeIndex.suffix, 'btree')

    def test_deconstruction(self):
        index = BTreeIndex(fields=['title'], name='test_title_btree', fillfactor=80)
        path, args, kwargs = index.deconstruct()
        self.assertEqual(path, 'django.contrib.postgres.indexes.BTreeIndex')
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {'fields': ['title'], 'name': 'test_title_btree', 'fillfactor': 80})

class GinIndexTests(IndexTestMixin, PostgreSQLSimpleTestCase):
    index_class = GinIndex

    def test_suffix(self):
        self.assertEqual(GinIndex.suffix, 'gin')

    def test_deconstruction(self):
        index = GinIndex(
            fields=['title'],
            name='test_title_gin',
            fastupdate=True,
            gin_pending_list_limit=128,
        )
        path, args, kwargs = index.deconstruct()
        self.assertEqual(path, 'django.contrib.postgres.indexes.GinIndex')
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {
            'fields': ['title'],
            'name': 'test_title_gin',
            'fastupdate': True,
            'gin_pending_list_limit': 128,
        })

class GistIndexTests(IndexTestMixin, PostgreSQLSimpleTestCase):
    index_class = GistIndex

    def test_suffix(self):
        self.assertEqual(GistIndex.suffix, 'gist')

    def test_deconstruction(self):
        index = GistIndex(fields=['title'], name='test_title_gist', buffering=False, fillfactor=80)
        path, args, kwargs = index.deconstruct()
        self.assertEqual(path, 'django.contrib.postgres.indexes.GistIndex')
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {
            'fields': ['title'],
            'name': 'test_title_gist',
            'buffering': False,
            'fillfactor': 80,
        })

class HashIndexTests(IndexTestMixin, PostgreSQLSimpleTestCase):
    index_class = HashIndex

    def test_suffix(self):
        self.assertEqual(HashIndex.suffix, 'hash')

    def test_deconstruction(self):
        index = HashIndex(fields=['title'], name='test_title_hash', fillfactor=80)
        path, args, kwargs = index.deconstruct()
        self.assertEqual(path, 'django.contrib.postgres.indexes.HashIndex')
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {'fields': ['title'], 'name': 'test_title_hash', 'fillfactor': 80})

class SpGistIndexTests(IndexTestMixin, PostgreSQLSimpleTestCase):
    index_class = SpGistIndex

    def test_suffix(self):
        self.assertEqual(SpGistIndex.suffix, 'spgist')

    def test_deconstruction(self):
        index = SpGistIndex(fields=['title'], name='test_title_spgist', fillfactor=80)
        path, args, kwargs = index.deconstruct()
        self.assertEqual(path, 'django.contrib.postgres.indexes.SpGistIndex')
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {'fields': ['title'], 'name': 'test_title_spgist', 'fillfactor': 80})

@modify_settings(INSTALLED_APPS={'append': 'django.contrib.postgres'})
class SchemaTests(PostgreSQLTestCase):
    get_opclass_query = '''
        SELECT opcname, c.relname FROM pg_opclass AS oc
        JOIN pg_index as i on oc.oid = ANY(i.indclass)
        JOIN pg_class as c on c.oid = i.indexrelid
        WHERE c.relname = %s
    '''

    def get_constraints(self, table):
        """
        Get the indexes on the table using a new cursor.
        """
        with connection.cursor() as cursor:
            return connection.introspection.get_constraints(cursor, table)

    def test_gin_index(self):
        # Ensure the table is there and doesn't have an index.
        self.assertNotIn('field', self.get_constraints(IntegerArrayModel._meta.db_table))
        # Add the index
        index_name = 'integer_array_model_field_gin'
        index = GinIndex(fields=['field'], name=index_name)
        with connection.schema_editor() as editor:
            editor.add_index(IntegerArrayModel, index)
        constraints = self.get_constraints(IntegerArrayModel._meta.db_table)
        # Check gin index was added
        self.assertEqual(constraints[index_name]['type'], GinIndex.suffix)
        # Drop the index
        with connection.schema_editor() as editor:
            editor.remove_index(IntegerArrayModel, index)
        self.assertNotIn(index_name, self.get_constraints(IntegerArrayModel._meta.db_table))

    def test_gin_fastupdate(self):
        index_name = 'integer_array_gin_fastupdate'
        index = GinIndex(fields=['field'], name=index_name, fastupdate=False)
        with connection.schema_editor() as editor:
            editor.add_index(IntegerArrayModel, index)
        constraints = self.get_constraints(IntegerArrayModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], 'gin')
        self.assertEqual(constraints[index_name]['options'], ['fastupdate=off'])
        with connection.schema_editor() as editor:
            editor.remove_index(IntegerArrayModel, index)
        self.assertNotIn(index_name, self.get_constraints(IntegerArrayModel._meta.db_table))

    def test_partial_gin_index(self):
        with register_lookup(CharField, Length):
            index_name = 'char_field_gin_partial_idx'
            index = GinIndex(fields=['field'], name=index_name, condition=Q(field__length=40))
            with connection.schema_editor() as editor:
                editor.add_index(CharFieldModel, index)
            constraints = self.get_constraints(CharFieldModel._meta.db_table)
            self.assertEqual(constraints[index_name]['type'], 'gin')
            with connection.schema_editor() as editor:
                editor.remove_index(CharFieldModel, index)
            self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_partial_gin_index_with_tablespace(self):
        with register_lookup(CharField, Length):
            index_name = 'char_field_gin_partial_idx'
            index = GinIndex(
                fields=['field'],
                name=index_name,
                condition=Q(field__length=40),
                db_tablespace='pg_default',
            )
            with connection.schema_editor() as editor:
                editor.add_index(CharFieldModel, index)
                self.assertIn('TABLESPACE "pg_default" ', str(index.create_sql(CharFieldModel, editor)))
            constraints = self.get_constraints(CharFieldModel._meta.db_table)
            self.assertEqual(constraints[index_name]['type'], 'gin')
            with connection.schema_editor() as editor:
                editor.remove_index(CharFieldModel, index)
            self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_gin_parameters(self):
        index_name = 'integer_array_gin_params'
        index = GinIndex(fields=['field'], name=index_name, fastupdate=True, gin_pending_list_limit=64)
        with connection.schema_editor() as editor:
            editor.add_index(IntegerArrayModel, index)
        constraints = self.get_constraints(IntegerArrayModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], 'gin')
        self.assertEqual(constraints[index_name]['options'], ['gin_pending_list_limit=64', 'fastupdate=on'])
        with connection.schema_editor() as editor:
            editor.remove_index(IntegerArrayModel, index)
        self.assertNotIn(index_name, self.get_constraints(IntegerArrayModel._meta.db_table))

    def test_trigram_op_class_gin_index(self):
        index_name = 'trigram_op_class_gin'
        index = GinIndex(OpClass(F('scene'), name='gin_trgm_ops'), name=index_name)
        with connection.schema_editor() as editor:
            editor.add_index(Scene, index)
        with editor.connection.cursor() as cursor:
            cursor.execute(self.get_opclass_query, [index_name])
            self.assertCountEqual(cursor.fetchall(), [('gin_trgm_ops', index_name)])
        constraints = self.get_constraints(Scene._meta.db_table)
        self.assertIn(index_name, constraints)
        self.assertIn(constraints[index_name]['type'], GinIndex.suffix)
        with connection.schema_editor() as editor:
            editor.remove_index(Scene, index)
        self.assertNotIn(index_name, self.get_constraints(Scene._meta.db_table))

    def test_cast_search_vector_gin_index(self):
        index_name = 'cast_search_vector_gin'
        index = GinIndex(Cast('field', SearchVectorField()), name=index_name)
        with connection.schema_editor() as editor:
            editor.add_index(TextFieldModel, index)
            sql = index.create_sql(TextFieldModel, editor)
        table = TextFieldModel._meta.db_table
        constraints = self.get_constraints(table)
        self.assertIn(index_name, constraints)
        self.assertIn(constraints[index_name]['type'], GinIndex.suffix)
        self.assertIs(sql.references_column(table, 'field'), True)
        self.assertIn(' AS tsvector', str(sql))
        with connection.schema_editor() as editor:
            editor.remove_index(TextFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(table))

    def test_bloom_index(self):
        index_name = 'char_field_model_field_bloom'
        index = BloomIndex(fields=['field'], name=index_name)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], BloomIndex.suffix)
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_bloom_parameters(self):
        index_name = 'char_field_model_field_bloom_params'
        index = BloomIndex(fields=['field'], name=index_name, length=512, columns=[3])
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], BloomIndex.suffix)
        self.assertEqual(constraints[index_name]['options'], ['length=512', 'col1=3'])
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_brin_index(self):
        index_name = 'char_field_model_field_brin'
        index = BrinIndex(fields=['field'], name=index_name, pages_per_range=4)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], BrinIndex.suffix)
        self.assertEqual(constraints[index_name]['options'], ['pages_per_range=4'])
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_brin_parameters(self):
        index_name = 'char_field_brin_params'
        index = BrinIndex(fields=['field'], name=index_name, autosummarize=True)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], BrinIndex.suffix)
        self.assertEqual(constraints[index_name]['options'], ['autosummarize=on'])
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_btree_index(self):
        # Ensure the table is there and doesn't have an index.
        self.assertNotIn('field', self.get_constraints(CharFieldModel._meta.db_table))
        # Add the index.
        index_name = 'char_field_model_field_btree'
        index = BTreeIndex(fields=['field'], name=index_name)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        # The index was added.
        self.assertEqual(constraints[index_name]['type'], BTreeIndex.suffix)
        # Drop the index.
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_btree_parameters(self):
        index_name = 'integer_array_btree_fillfactor'
        index = BTreeIndex(fields=['field'], name=index_name, fillfactor=80)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], BTreeIndex.suffix)
        self.assertEqual(constraints[index_name]['options'], ['fillfactor=80'])
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_gist_index(self):
        # Ensure the table is there and doesn't have an index.
        self.assertNotIn('field', self.get_constraints(CharFieldModel._meta.db_table))
        # Add the index.
        index_name = 'char_field_model_field_gist'
        index = GistIndex(fields=['field'], name=index_name)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        # The index was added.
        self.assertEqual(constraints[index_name]['type'], GistIndex.suffix)
        # Drop the index.
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_gist_parameters(self):
        index_name = 'integer_array_gist_buffering'
        index = GistIndex(fields=['field'], name=index_name, buffering=True, fillfactor=80)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], GistIndex.suffix)
        self.assertEqual(constraints[index_name]['options'], ['buffering=on', 'fillfactor=80'])
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    @skipUnlessDBFeature('supports_covering_gist_indexes')
    def test_gist_include(self):
        index_name = 'scene_gist_include_setting'
        index = GistIndex(name=index_name, fields=['scene'], include=['setting'])
        with connection.schema_editor() as editor:
            editor.add_index(Scene, index)
        constraints = self.get_constraints(Scene._meta.db_table)
        self.assertIn(index_name, constraints)
        self.assertEqual(constraints[index_name]['type'], GistIndex.suffix)
        self.assertEqual(constraints[index_name]['columns'], ['scene', 'setting'])
        with connection.schema_editor() as editor:
            editor.remove_index(Scene, index)
        self.assertNotIn(index_name, self.get_constraints(Scene._meta.db_table))

    def test_gist_include_not_supported(self):
        index_name = 'gist_include_exception'
        index = GistIndex(fields=['scene'], name=index_name, include=['setting'])
        msg = 'Covering GiST indexes requires PostgreSQL 12+.'
        with self.assertRaisesMessage(NotSupportedError, msg):
            with mock.patch(
                'django.db.backends.postgresql.features.DatabaseFeatures.supports_covering_gist_indexes',
                False,
            ):
                with connection.schema_editor() as editor:
                    editor.add_index(Scene, index)
        self.assertNotIn(index_name, self.get_constraints(Scene._meta.db_table))

    def test_tsvector_op_class_gist_index(self):
        index_name = 'tsvector_op_class_gist'
        index = GistIndex(
            OpClass(
                SearchVector('scene', 'setting', config='english'),
                name='tsvector_ops',
            ),
            name=index_name,
        )
        with connection.schema_editor() as editor:
            editor.add_index(Scene, index)
            sql = index.create_sql(Scene, editor)
        table = Scene._meta.db_table
        constraints = self.get_constraints(table)
        self.assertIn(index_name, constraints)
        self.assertIn(constraints[index_name]['type'], GistIndex.suffix)
        self.assertIs(sql.references_column(table, 'scene'), True)
        self.assertIs(sql.references_column(table, 'setting'), True)
        with connection.schema_editor() as editor:
            editor.remove_index(Scene, index)
        self.assertNotIn(index_name, self.get_constraints(table))

    def test_hash_index(self):
        # Ensure the table is there and doesn't have an index.
        self.assertNotIn('field', self.get_constraints(CharFieldModel._meta.db_table))
        # Add the index.
        index_name = 'char_field_model_field_hash'
        index = HashIndex(fields=['field'], name=index_name)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        # The index was added.
        self.assertEqual(constraints[index_name]['type'], HashIndex.suffix)
        # Drop the index.
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_hash_parameters(self):
        index_name = 'integer_array_hash_fillfactor'
        index = HashIndex(fields=['field'], name=index_name, fillfactor=80)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], HashIndex.suffix)
        self.assertEqual(constraints[index_name]['options'], ['fillfactor=80'])
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_spgist_index(self):
        # Ensure the table is there and doesn't have an index.
        self.assertNotIn('field', self.get_constraints(CharFieldModel._meta.db_table))
        # Add the index.
        index_name = 'char_field_model_field_spgist'
        index = SpGistIndex(fields=['field'], name=index_name)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        # The index was added.
        self.assertEqual(constraints[index_name]['type'], SpGistIndex.suffix)
        # Drop the index.
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_spgist_parameters(self):
        index_name = 'integer_array_spgist_fillfactor'
        index = SpGistIndex(fields=['field'], name=index_name, fillfactor=80)
        with connection.schema_editor() as editor:
            editor.add_index(CharFieldModel, index)
        constraints = self.get_constraints(CharFieldModel._meta.db_table)
        self.assertEqual(constraints[index_name]['type'], SpGistIndex.suffix)
        self.assertEqual(constraints[index_name]['options'], ['fillfactor=80'])
        with connection.schema_editor() as editor:
            editor.remove_index(CharFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(CharFieldModel._meta.db_table))

    def test_op_class(self):
        index_name = 'test_op_class'
        index = Index(
            OpClass(Lower('field'), name='text_pattern_ops'),
            name=index_name,
        )
        with connection.schema_editor() as editor:
            editor.add_index(TextFieldModel, index)
        with editor.connection.cursor() as cursor:
            cursor.execute(self.get_opclass_query, [index_name])
            self.assertCountEqual(cursor.fetchall(), [('text_pattern_ops', index_name)])

    def test_op_class_descending_collation(self):
        collation = connection.features.test_collations.get('non_default')
        if not collation:
            self.skipTest(
                'This backend does not support case-insensitive collations.'
            )
        index_name = 'test_op_class_descending_collation'
        index = Index(
            Collate(
                OpClass(Lower('field'), name='text_pattern_ops').desc(nulls_last=True),
                collation=collation,
            ),
            name=index_name,
        )
        with connection.schema_editor() as editor:
            editor.add_index(TextFieldModel, index)
            self.assertIn(
                'COLLATE %s' % editor.quote_name(collation),
                str(index.create_sql(TextFieldModel, editor)),
            )
        with editor.connection.cursor() as cursor:
            cursor.execute(self.get_opclass_query, [index_name])
            self.assertCountEqual(cursor.fetchall(), [('text_pattern_ops', index_name)])
        table = TextFieldModel._meta.db_table
        constraints = self.get_constraints(table)
        self.assertIn(index_name, constraints)
        self.assertEqual(constraints[index_name]['orders'], ['DESC'])
        with connection.schema_editor() as editor:
            editor.remove_index(TextFieldModel, index)
        self.assertNotIn(index_name, self.get_constraints(table))

    def test_op_class_descending_partial(self):
        index_name = 'test_op_class_descending_partial'
        index = Index(
            OpClass(Lower('field'), name='text_pattern_ops').desc(),
            name=index_name,
            condition=Q(field__contains='China'),
        )
        with connection.schema_editor() as editor:
            editor.add_index(TextFieldModel, index)
        with editor.connection.cursor() as cursor:
            cursor.execute(self.get_opclass_query, [index_name])
            self.assertCountEqual(cursor.fetchall(), [('text_pattern_ops', index_name)])
        constraints = self.get_constraints(TextFieldModel._meta.db_table)
        self.assertIn(index_name, constraints)
        self.assertEqual(constraints[index_name]['orders'], ['DESC'])

    def test_op_class_descending_partial_tablespace(self):
        index_name = 'test_op_class_descending_partial_tablespace'
        index = Index(
            OpClass(Lower('field').desc(), name='text_pattern_ops'),
            name=index_name,
            condition=Q(field__contains='China'),
            db_tablespace='pg_default',
        )
        with connection.schema_editor() as editor:
            editor.add_index(TextFieldModel, index)
            self.assertIn(
                'TABLESPACE "pg_default" ',
                str(index.create_sql(TextFieldModel, editor))
            )
        with editor.connection.cursor() as cursor:
            cursor.execute(self.get_opclass_query, [index_name])
            self.assertCountEqual(cursor.fetchall(), [('text_pattern_ops', index_name)])
        constraints = self.get_constraints(TextFieldModel._meta.db_table)
        self.assertIn(index_name, constraints)
        self.assertEqual(constraints[index_name]['orders'], ['DESC'])


text:
@interface NullView : UIView

@property (nonatomic, copy) NSString *nullIconImageName;

@property (nonatomic, copy) NSString *desText;

@end

redpajama_set_name: RedPajamaGithub


text:
use std::ops::{Deref, DerefMut};

use Resolver;
use Namespace::{TypeNS, ValueNS};

use rustc::lint;
use rustc::middle::privacy::{DependsOn, LastImport, Used, Unused};
use syntax::ast;
use syntax::ast::{ViewItem, ViewItemExternCrate, ViewItemUse};
use syntax::ast::{ViewPathGlob, ViewPathList, ViewPathSimple};
use syntax::codemap::{Span, DUMMY_SP};
use syntax::visit::{self, Visitor};

struct UnusedImportCheckVisitor<'a, 'b:'a, 'tcx:'b> {
    resolver: &'a mut Resolver<'b, 'tcx>
}

// Deref and DerefMut impls allow treating UnusedImportCheckVisitor as Resolver.
impl<'a, 'b, 'tcx:'b> Deref for UnusedImportCheckVisitor<'a, 'b, 'tcx> {
    type Target = Resolver<'b, 'tcx>;

    fn deref<'c>(&'c self) -> &'c Resolver<'b, 'tcx> {
        &*self.resolver
    }
}

impl<'a, 'b, 'tcx:'b> DerefMut for UnusedImportCheckVisitor<'a, 'b, 'tcx> {
    fn deref_mut<'c>(&'c mut self) -> &'c mut Resolver<'b, 'tcx> {
        &mut *self.resolver
    }
}

impl<'a, 'b, 'tcx> UnusedImportCheckVisitor<'a, 'b, 'tcx> {
    // We have information about whether `use` (import) directives are actually used now.
    // If an import is not used at all, we signal a lint error. If an import is only used
    // for a single namespace, we remove the other namespace from the recorded privacy
    // information. That means in privacy.rs, we will only check imports and namespaces
    // which are used. In particular, this means that if an import could name either a
    // public or private item, we will check the correct thing, dependent on how the import
    // is used.
    fn finalize_import(&mut self, id: ast::NodeId, span: Span) {
        debug!("finalizing import uses for {:?}",
                self.session.codemap().span_to_snippet(span));

        if !self.used_imports.contains(&(id, TypeNS)) &&
           !self.used_imports.contains(&(id, ValueNS)) {
            self.session.add_lint(lint::builtin::UNUSED_IMPORTS,
                                  id,
                                  span,
                                  "unused import".to_string());
        }

        let (v_priv, t_priv) = match self.last_private.get(&id) {
            Some(&LastImport {
                value_priv: v,
                value_used: _,
                type_priv: t,
                type_used: _
            }) => (v, t),
            Some(_) => {
                panic!("we should only have LastImport for `use` directives")
            }
            _ => return,
        };

        let mut v_used = if self.used_imports.contains(&(id, ValueNS)) {
            Used
        } else {
            Unused
        };
        let t_used = if self.used_imports.contains(&(id, TypeNS)) {
            Used
        } else {
            Unused
        };

        match (v_priv, t_priv) {
            // Since some items may be both in the value _and_ type namespaces (e.g., structs)
            // we might have two LastPrivates pointing at the same thing. There is no point
            // checking both, so lets not check the value one.
            (Some(DependsOn(def_v)), Some(DependsOn(def_t))) if def_v == def_t => v_used = Unused,
            _ => {},
        }

        self.last_private.insert(id, LastImport{value_priv: v_priv,
                                                value_used: v_used,
                                                type_priv: t_priv,
                                                type_used: t_used});
    }
}

impl<'a, 'b, 'v, 'tcx> Visitor<'v> for UnusedImportCheckVisitor<'a, 'b, 'tcx> {
    fn visit_view_item(&mut self, vi: &ViewItem) {
        // Ignore is_public import statements because there's no way to be sure
        // whether they're used or not. Also ignore imports with a dummy span
        // because this means that they were generated in some fashion by the
        // compiler and we don't need to consider them.
        if vi.vis == ast::Public || vi.span == DUMMY_SP {
            visit::walk_view_item(self, vi);
            return;
        }

        match vi.node {
            ViewItemExternCrate(_, _, id) => {
                if let Some(crate_num) = self.session.cstore.find_extern_mod_stmt_cnum(id) {
                    if !self.used_crates.contains(&crate_num) {
                        self.session.add_lint(lint::builtin::UNUSED_EXTERN_CRATES,
                                              id,
                                              vi.span,
                                              "unused extern crate".to_string());
                    }
                }
            },
            ViewItemUse(ref p) => {
                match p.node {
                    ViewPathSimple(_, _, id) => {
                        self.finalize_import(id, p.span)
                    }

                    ViewPathList(_, ref list, _) => {
                        for i in list.iter() {
                            self.finalize_import(i.node.id(), i.span);
                        }
                    }
                    ViewPathGlob(_, id) => {
                        if !self.used_imports.contains(&(id, TypeNS)) &&
                           !self.used_imports.contains(&(id, ValueNS)) {
                            self.session
                                .add_lint(lint::builtin::UNUSED_IMPORTS,
                                          id,
                                          p.span,
                                          "unused import".to_string());
                        }
                    }
                }
            }
        }

        visit::walk_view_item(self, vi);
    }
}

pub fn check_crate(resolver: &mut Resolver, krate: &ast::Crate) {
    let mut visitor = UnusedImportCheckVisitor { resolver: resolver };
    visit::walk_crate(&mut visitor, krate);
}


text:
<?php

class Payeezy
{
  /**
   * @var string The Payeezy API params to be used for requests.
   */
  public static $apiKey;
  public static $apiSecret;
  public static $merchantToken;
  public static $baseURL;
  public static $url;

  /**
   * Sets the API key to be used for requests.
   *
   * @param string $apiKey
   */
  public static function setApiKey($apiKey)
  {
    self::$apiKey = $apiKey;
  }
  /**
   * Sets the API key to be used for requests.
   *
   * @param string $apiKey
   */
  public static function getApiKey()
  {
    return self::$apiKey;
  }
    /**
   * Sets the API secret to be used for requests.
   *
   * @param string $apiSecret
   */
  public static function setApiSecret($apiSecret)
  {
    self::$apiSecret = $apiSecret;
  }
  /**
   * Sets the API secret to be used for requests.
   *
   * @param string $apiSecret
   */
  public static function getApiSecret()
  {
    return self::$apiSecret;
  }
  /**
   * Sets the API Base URL.
   *
   * @param string $url
   */
  public static function setUrl($baseURL)
  {
    self::$baseURL = $baseURL;
  }
  /**
   * Gets the API Base URL.
   *
   * @param string $url
   */
  public static function getUrl()
  {
    return self::$baseURL;
  }
  /**
   * Sets the API Merchant Token
   *
   * @param string $merchantToken
   */
  public static function setMerchantToken($merchantToken)
  {
    self::$merchantToken = $merchantToken;
  }
  /**
   * Gets the API Merchant Token
   *
   * @param string $merchantToken
   */
  public static function getMerchantToken()
  {
    return self::$merchantToken;
  }

  /**
   * Payeezy
   *
   * Generate Payload
   */

  public function getPayload($args = array())
  {
    $args = array_merge(array(
        "amount"=> "",
        "card_number" => "",
        "card_type" => "",
        "card_holder_name" => "",
        "card_cvv" => "",
        "card_expiry" => "",
        "merchant_ref" => "",
        "currency_code" => "",
        "transaction_tag" => "",
        "split_shipment" => "",
        "transaction_id" => "",

    ), $args);

    $transaction_type = strtolower(func_get_arg(1));

    $data = "";
    if($transaction_type == "authorize" || $transaction_type == "purchase")
    {
      $data = array(
              'merchant_ref'=> $args['merchant_ref'],
              'transaction_type'=> $transaction_type,
              'method'=> 'credit_card',
              'amount'=> $args['amount'],
              'currency_code'=> strtoupper($args['currency_code']),
              'credit_card'=> array(
                      'type'=> $args['card_type'],
                      'cardholder_name'=> $args['card_holder_name'],
                      'card_number'=> $args['card_number'],
                      'exp_date'=> $args['card_expiry'],
                      'cvv'=> $args['card_cvv'],
                    )
      );

      self::$url = self::$baseURL;
    }else{

      self::$url = self::$baseURL . '/' . $args['transaction_id'];

      if($transaction_type == "split")
      {
        $data = array(
          'merchant_ref'=> $args['merchant_ref'],
          'transaction_type'=> $transaction_type,
          'method'=> 'credit_card',
          'amount'=> $args['amount'],
          'currency_code'=> strtoupper($args['currency_code']),
          'transaction_tag'=>$args['transaction_tag'],
          'split_shipment'=>$args['split_shipment'],
        );

      }else{
        $data = array(
                'merchant_ref'=> $args['merchant_ref'],
                'transaction_type'=> $transaction_type,
                'method'=> 'credit_card',
                'amount'=> $args['amount'],
                'currency_code'=> strtoupper($args['currency_code']),
                'transaction_tag'=>$args['transaction_tag'],
        );

      }
    }
    return json_encode($data, JSON_FORCE_OBJECT);
  }

  /**
   * Payeezy
   *
   * HMAC Authentication
   */

  public function hmacAuthorizationToken($payload)
  {

    $nonce = strval(hexdec(bin2hex(openssl_random_pseudo_bytes(4, $cstrong))));

    $timestamp = strval(time()*1000); //time stamp in milli seconds

    $data = self::$apiKey . $nonce . $timestamp . self::$merchantToken . $payload;

    $hashAlgorithm = "sha256";

    $hmac = hash_hmac ( $hashAlgorithm , $data , self::$apiSecret, false );    // HMAC Hash in hex

    $authorization = base64_encode($hmac);

    return array(
            'authorization' => $authorization,
            'nonce' => $nonce,
            'timestamp' => $timestamp,
      );
  }

  /**
   * jsonpp - Pretty print JSON data
   *
   * In versions of PHP < 5.4.x, the json_encode() function does not yet provide a
   * pretty-print option. In lieu of forgoing the feature, an additional call can
   * be made to this function, passing in JSON text, and (optionally) a string to
   * be used for indentation.
   *
   * @param string $json  The JSON data, pre-encoded
   * @param string $istr  The indentation string
   *
   * @return string
   */
  public function jsonpp($json, $istr='  ')
  {
      $result = '';
      for($p=$q=$i=0; isset($json[$p]); $p++)
      {
          $json[$p] == '"' && ($p>0?$json[$p-1]:'') != '\\' && $q=!$q;
          if(strchr('}]', $json[$p]) && !$q && $i--)
          {
              strchr('{[', $json[$p-1]) || $result .= "\n".str_repeat($istr, $i);
          }
          $result .= $json[$p];
          if(strchr(',{[', $json[$p]) && !$q)
          {
              $i += strchr('{[', $json[$p])===FALSE?0:1;
              strchr('}]', $json[$p+1]) || $result .= "\n".str_repeat($istr, $i);
          }
      }
      return $result;
  }

  /**
   * Payeezy
   *
   * Post Transaction
   */

  public  function postTransaction($payload, $headers)
  {

    $request = curl_init();
    curl_setopt($request, CURLOPT_URL, self::$url );
    curl_setopt($request, CURLOPT_POST, true);
    curl_setopt($request, CURLOPT_POSTFIELDS, $payload);
    curl_setopt($request, CURLOPT_RETURNTRANSFER, true);
    curl_setopt($request, CURLOPT_HEADER, false);
	//curl_setopt($request, CURLOPT_SSL_VERIFYPEER, false);
    curl_setopt($request, CURLOPT_HTTPHEADER, array(
      'Content-Type: application/json',
      'apikey:'.strval(self::$apiKey),
      'token:'.strval(self::$merchantToken),
      'Authorization:'.$headers['authorization'],
      'nonce:'.$headers['nonce'],
      'timestamp:'.$headers['timestamp'],
    ));

    $response = curl_exec($request);

	if (FALSE === $response)
        echo curl_error($request);

    //$httpcode = curl_getinfo($request, CURLINFO_HTTP_CODE);
    curl_close($request);

    return $response;
  }

  /**
   * Payeezy

   *
   * Authorize Transaction
   */

  public function authorize($args = array())
  {
      $payload = $this->getPayload($args, "authorize");

      $headerArray = $this->hmacAuthorizationToken($payload);

      return $this->postTransaction($payload, $headerArray);
  }

  /**
   * Payeezy
   *
   * Purchase Transaction
   */

  public function purchase($args = array())
  {
      $payload = $this->getPayload($args, "purchase");
      $headerArray = $this->hmacAuthorizationToken($payload);
      return $this->postTransaction($payload, $headerArray);
  }

  /**
   * Payeezy
   *
   * Capture Transaction
   */

  public function capture($args = array())
  {
      $payload = $this->getPayload($args, "capture");
      $headerArray = $this->hmacAuthorizationToken($payload);
      return $this->postTransaction($payload, $headerArray);
  }

  /**
   * Payeezy
   *
   * Void Transaction
   */

  public function void($args = array())
  {
      $payload = $this->getPayload($args, "void");
      $headerArray = $this->hmacAuthorizationToken($payload);
      return $this->postTransaction($payload, $headerArray);
  }

  /**
   * Payeezy
   *
   * Refund Transaction
   */

  public function refund($args = array())
  {
      $payload = $this->getPayload($args, "refund");
      $headerArray = $this->hmacAuthorizationToken($payload);
      return $this->postTransaction($payload, $headerArray);
  }

  /**
   * Payeezy
   *
   * split Transaction
   */

  public function split_shipment($args = array())
  {
      $payload = $this->getPayload($args, "split");
      $headerArray = $this->hmacAuthorizationToken($payload);
      return $this->postTransaction($payload, $headerArray);
  }

}


text:
package internal

import (
	"log"
	"sync"

	"github.com/AutoRoute/node/types"
)

// Takes care of handling packet receipts, namely relaying them to other
// interested hosts and sending them to any objects which want to take action
// on them via the ReceiptAction interface.
type receiptHandler struct {
	connections map[types.NodeAddress]ReceiptConnection
	packets     map[types.PacketHash]routingDecision
	l           *sync.Mutex
	id          types.NodeAddress
	outgoing    chan types.PacketHash
	logger      Logger
	quit        chan bool
}

func newReceipt(id types.NodeAddress, c <-chan routingDecision, route_logger Logger) *receiptHandler {
	r := &receiptHandler{
		make(map[types.NodeAddress]ReceiptConnection),
		make(map[types.PacketHash]routingDecision),
		&sync.Mutex{},
		id,
		make(chan types.PacketHash),
		route_logger,
		make(chan bool),
	}
	go r.sentPackets(c)
	return r
}

func (r *receiptHandler) AddConnection(id types.NodeAddress, c ReceiptConnection) {
	r.l.Lock()
	r.connections[id] = c
	r.l.Unlock()
	go r.handleConnection(id, c)
}

func (r *receiptHandler) handleConnection(id types.NodeAddress, c ReceiptConnection) {
	for {
		select {
		case receipt, ok := <-c.PacketReceipts():
			if !ok {
				return
			}
			r.sendReceipt(id, receipt)
		case <-r.quit:
			return
		}
	}
}

func (r *receiptHandler) sentPackets(c <-chan routingDecision) {
	for {
		select {
		case d := <-c:
			r.l.Lock()
			r.packets[d.hash] = d
			r.l.Unlock()
		case <-r.quit:
			return
		}
	}
}

func (r *receiptHandler) PacketHashes() <-chan types.PacketHash {
	return r.outgoing
}

func (r *receiptHandler) SendReceipt(receipt PacketReceipt) {
	r.sendReceipt(r.id, receipt)
}

func (r *receiptHandler) sendReceipt(id types.NodeAddress, receipt PacketReceipt) {
	if err := receipt.Verify(); err != nil {
		log.Printf("Error verifying receipt from %x: %q", id, err)
		log.Print(receipt)
		return
	}
	dest := make(map[types.NodeAddress]bool)
	r.l.Lock()
	defer r.l.Unlock()
	for _, hash := range receipt.ListPackets() {
		record, ok := r.packets[hash]
		if !ok {
			log.Printf("No record found %q", hash)
			continue
		}
		if record.destination != receipt.Source() {
			log.Printf("Invalid source %q != %q", record.source, receipt.Source())
			continue
		}
		if record.nexthop != id {
			log.Printf("Received packet receipt from wrong host? %q != %q", id, record.nexthop)
		}
		dest[record.source] = true
		r.outgoing <- hash

		err := r.logger.LogPacketReceipt(hash)
		if err != nil {
			log.Fatal("Couldn't log receipt?")
		}
	}
	for addr, _ := range dest {
		if addr != r.id {
			r.connections[addr].SendReceipt(receipt)
		}
	}
}

func (r *receiptHandler) Close() error {
	close(r.quit)
	return nil
}


text:
// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.
// Code generated by Microsoft (R) AutoRest Code Generator.

package com.azure.resourcemanager.avs.fluent;

import com.azure.core.annotation.ReturnType;
import com.azure.core.annotation.ServiceMethod;
import com.azure.core.http.rest.PagedIterable;
import com.azure.core.http.rest.Response;
import com.azure.core.util.Context;
import com.azure.resourcemanager.avs.fluent.models.ScriptPackageInner;

/** An instance of this class provides access to all the operations defined in ScriptPackagesClient. */
public interface ScriptPackagesClient {
    /**
     * List script packages available to run on the private cloud.
     *
     * @param resourceGroupName The name of the resource group. The name is case insensitive.
     * @param privateCloudName Name of the private cloud.
     * @throws IllegalArgumentException thrown if parameters fail the validation.
     * @throws com.azure.core.management.exception.ManagementException thrown if the request is rejected by server.
     * @throws RuntimeException all other wrapped checked exceptions if the request fails to be sent.
     * @return a list of the available script packages as paginated response with {@link PagedIterable}.
     */
    @ServiceMethod(returns = ReturnType.COLLECTION)
    PagedIterable<ScriptPackageInner> list(String resourceGroupName, String privateCloudName);

    /**
     * List script packages available to run on the private cloud.
     *
     * @param resourceGroupName The name of the resource group. The name is case insensitive.
     * @param privateCloudName Name of the private cloud.
     * @param context The context to associate with this operation.
     * @throws IllegalArgumentException thrown if parameters fail the validation.
     * @throws com.azure.core.management.exception.ManagementException thrown if the request is rejected by server.
     * @throws RuntimeException all other wrapped checked exceptions if the request fails to be sent.
     * @return a list of the available script packages as paginated response with {@link PagedIterable}.
     */
    @ServiceMethod(returns = ReturnType.COLLECTION)
    PagedIterable<ScriptPackageInner> list(String resourceGroupName, String privateCloudName, Context context);

    /**
     * Get a script package available to run on a private cloud.
     *
     * @param resourceGroupName The name of the resource group. The name is case insensitive.
     * @param privateCloudName Name of the private cloud.
     * @param scriptPackageName Name of the script package in the private cloud.
     * @param context The context to associate with this operation.
     * @throws IllegalArgumentException thrown if parameters fail the validation.
     * @throws com.azure.core.management.exception.ManagementException thrown if the request is rejected by server.
     * @throws RuntimeException all other wrapped checked exceptions if the request fails to be sent.
     * @return a script package available to run on a private cloud along with {@link Response}.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    Response<ScriptPackageInner> getWithResponse(
        String resourceGroupName, String privateCloudName, String scriptPackageName, Context context);

    /**
     * Get a script package available to run on a private cloud.
     *
     * @param resourceGroupName The name of the resource group. The name is case insensitive.
     * @param privateCloudName Name of the private cloud.
     * @param scriptPackageName Name of the script package in the private cloud.
     * @throws IllegalArgumentException thrown if parameters fail the validation.
     * @throws com.azure.core.management.exception.ManagementException thrown if the request is rejected by server.
     * @throws RuntimeException all other wrapped checked exceptions if the request fails to be sent.
     * @return a script package available to run on a private cloud.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    ScriptPackageInner get(String resourceGroupName, String privateCloudName, String scriptPackageName);
}


text:
class TreeNode:
    def __init__(self, val=0, left=None, right=None):
        self.val = val
        self.left = left
        self.right = right

from tkinter.tix import Tree
from typing import Optional

class Solution:
    def inorderTraversal(self, root: Optional[TreeNode]) -> list[int]:
        # left->root->right
        result = []
        self.inorder(root, result)
        return result

    def inorder(self, root: Optional[TreeNode], result: list[int]) -> None:
        if root != None:
            self.inorder(root.left, result)
            result.append(root.val)
            self.inorder(root.right, result)

    def preorderTraversal(self, root: Optional[TreeNode]) -> list[int]:
        result = []
        self.preorder(root, result)
        return result

    def preorder(self, root: Optional[TreeNode], result: list[int]) -> None:
        if root != None:
            result.append(root.val)
            self.preorder(root.left, result)
            self.preorder(root.right, result)

    def postorderTraversal(self, root: Optional[TreeNode]) -> list[int]:
        result = []
        self.postorder(root, result)
        return result

    def postorder(self, root: Optional[TreeNode], result: list[int]) -> None:
        if root != None:
            self.postorder(root.left, result)
            self.postorder(root.right, result)
            result.append(root.val)

one = TreeNode(1)
two = TreeNode(2)
three = TreeNode(3)
one.left = two
one.right = three
four = TreeNode(4)
two.left = four
print(Solution().inorderTraversal(one))
print(Solution().preorderTraversal(one))


text:
using System;
using Penneo;
using System.Collections.Generic;
using System.Linq;
using System.IO;
using System.Diagnostics;

namespace Penneo
{
    public class DownloadSignedDocuments
    {
        public static void Main(string[] args)
        {
            if (args.Length != 4) {
                Console.WriteLine("Parameters required: endpoint, key, secret, case-file-id");
                Environment.Exit(-1);
            }

            string endpoint = args[0];
            string key      = args[1];
            string secret   = args[2];

            int caseFileId  = System.Convert.ToInt32(args[3]);

            PenneoConnector.Initialize(key, secret, endpoint);
            PenneoConnector.SetLogger(new Logger());
            run(caseFileId);
        }

        public static void run(int caseFileId)
        {
            var caseFile = Query.Find<CaseFile>(caseFileId);

            // Check empty response
            if (caseFile == null) {
                Console.WriteLine("Nothing found");
                return;
            }

            Console.WriteLine("Case File Id: " + caseFile.Id);

            var documents = caseFile.GetDocuments();

            foreach (Document document in documents)
            {
                Console.WriteLine("- " + document.Id + " : " + document.Title + "(" + document.GetStatus() + ")");
                if (document.GetStatus().Equals("completed")) {
                    File.WriteAllBytes("downloaded-" + document.Id + ".pdf", document.GetPdf());
                }
            }
        }
    }

    internal class Logger : ILogger
    {
        public void Log(string message, LogSeverity severity)
        {
            if (severity >= LogSeverity.Error)
            {
                Debug.WriteLine(severity + ": " + message);
            }
        }
    }
}


text:
from signals.main import run_signals

__all__ = ['run_signals']


text:
package org.dasein.cloud.azurepack;

import org.apache.http.client.methods.RequestBuilder;

/**
 * Created by vmunthiu on 2/20/2015.
 */
public class AzurePackDataCenterRequests {

    private final String CLOUD_RESOURCES = "%s/%s/services/systemcenter/vmm/Clouds";
    private final String STAMP_RESOURCES = "";

    private AzurePackCloud provider;

    public AzurePackDataCenterRequests(AzurePackCloud provider){
        this.provider = provider;
    }

    public RequestBuilder listClouds(){
        RequestBuilder requestBuilder = RequestBuilder.get();
        addCommonHeaders(requestBuilder);
        requestBuilder.setUri(String.format(CLOUD_RESOURCES, this.provider.getContext().getEndpoint(), this.provider.getContext().getAccountNumber()));
        return requestBuilder;
    }

    private void addCommonHeaders(RequestBuilder requestBuilder) {
        requestBuilder.addHeader("x-ms-version", "2014-02-01");
        requestBuilder.addHeader("Accept", "application/json");
    }
}


text:
import json

import os
from faker import Faker
from jsonschema import ValidationError
from mock import mock
from pyfakefs.fake_filesystem_unittest import Patcher, TestCase

from mdstudio.api.exception import RegisterException
from mdstudio.api.schema import ISchema, ResourceSchema, EndpointSchema, HttpsSchema, InlineSchema, ClaimSchema, validate_json_schema, \
    MDStudioClaimSchema
from mdstudio.deferred.chainable import test_chainable
from mdstudio.unittest.db import DBTestCase

# python 2 compat
try:
    FileNotFoundError
except NameError:
    FileNotFoundError = IOError

class ISchemaTests(DBTestCase):
    faker = Faker()

    def test_construction(self):
        schema = ISchema()
        self.assertEqual(schema.cached, {})

    def test_retrieve_local(self):
        for i in range(50):
            with Patcher() as patcher:
                base_path = self.faker.file_path()
                schema_path = self.faker.file_path()
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.json', contents=json.dumps(content))

                schema = ISchema()
                schema._retrieve_local(base_path, schema_path)
                self.assertEqual(schema.cached, {
                    1: content
                })

    def test_retrieve_local_not_exists(self):
        for i in range(50):
            with Patcher() as patcher:
                base_path = self.faker.file_path()
                schema_path = self.faker.file_path()

                schema = ISchema()
                self.assertRaises(FileNotFoundError, schema._retrieve_local, base_path, schema_path)

    def test_retrieve_local_version(self):
        for i in range(50):
            with Patcher() as patcher:

                base_path = self.faker.file_path()
                schema_path = self.faker.file_path()
                versions = {self.faker.random_number(3) for _ in range(self.faker.random_number(2))}
                versions.add(self.faker.random_number(3))
                contents = {}
                for v in versions:
                    content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                    patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v{}.json'.format(v), contents=json.dumps(content))
                    contents[v] = content

                schema = ISchema()
                schema._retrieve_local(base_path, schema_path, versions=versions)
                self.assertEqual(schema.cached, contents)

    def test_retrieve_local_version_single(self):
        for i in range(50):
            with Patcher() as patcher:
                base_path = self.faker.file_path()
                schema_path = self.faker.file_path()
                version = self.faker.random_number(3)
                contents = {}
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v{}.json'.format(version), contents=json.dumps(content))
                contents[version] = content

                schema = ISchema()
                schema._retrieve_local(base_path, schema_path, versions=version)
                self.assertEqual(schema.cached, contents)

    def test_retrieve_local_version_single0(self):
        for i in range(50):
            with Patcher() as patcher:
                base_path = self.faker.file_path()
                schema_path = self.faker.file_path()
                version = 0
                contents = {}
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v{}.json'.format(version), contents=json.dumps(content))
                contents[version] = content

                schema = ISchema()
                schema._retrieve_local(base_path, schema_path, versions=version)
                self.assertEqual(schema.cached, contents)

    @test_chainable
    def test_recurse_subschemas_no_refs(self):

        for i in range(50):
            content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
            schema = ISchema()
            out = yield schema._recurse_subschemas(content, None)
            self.assertEqual(schema.cached, {})
            self.assertEqual(out, {
                'schema': content,
                'success': True
            })

    @test_chainable
    def test_recurse_subschemas_single_ref(self):

        for i in range(50):
            content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
            content['$ref'] = 'http://' + self.faker.domain_name()
            schema = ISchema()
            out = yield schema._recurse_subschemas(content, None)
            self.assertEqual(schema.cached, {})
            self.assertEqual(out, {
                'schema': content,
                'success': True
            })

    @test_chainable
    def test_recurse_subschemas_faulty_ref(self):

        for i in range(50):
            content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
            content['$ref'] = self.faker.domain_name()
            schema = ISchema()
            yield self.assertFailure(schema._recurse_subschemas(content, None), RegisterException)

    @test_chainable
    def test_recurse_subschemas_list_ref(self):

        for i in range(50):
            contents = []
            for _ in range(1 + self.faker.random_number(1)):
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                content['$ref'] = 'http://' + self.faker.domain_name()
                contents.append(content)

            schema = ISchema()
            out = yield schema._recurse_subschemas(contents, None)
            self.assertEqual(schema.cached, {})
            self.assertEqual(out, {
                'schema': contents,
                'success': True
            })

    @test_chainable
    def test_recurse_subschemas_endpoint(self):

        for i in range(5):

            with Patcher() as patcher:
                base_path = self.faker.file_path()

                base_contents = []
                for _ in range(1 + self.faker.random_number(1)):
                    versions = {self.faker.random_number(3) for _ in range(self.faker.random_number(1))}
                    versions.add(self.faker.random_number(3))
                    for v in versions:
                        content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                        schema_path = self.faker.word()

                        file = os.path.join(base_path, 'endpoints', schema_path) + '.v{}.json'.format(v)
                        if not os.path.isfile(file):
                            patcher.fs.CreateFile(file, contents=json.dumps(content))

                        base_content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                        base_content['$ref'] = 'endpoint://' + schema_path + '/v{}'.format(v)
                        base_contents.append(base_content)

                schema = ISchema()
                session = mock.MagicMock()
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                out = yield schema._recurse_subschemas(base_contents, session)
                self.assertEqual(schema.cached, {})
                self.assertEqual(out, {
                    'schema': base_contents,
                    'success': True
                })

    @test_chainable
    def test_recurse_subschemas_endpoint_fails(self):

        for i in range(5):

            with Patcher() as patcher:
                base_path = self.faker.file_path()

                base_contents = []
                for _ in range(1 + self.faker.random_number(1)):
                    versions = {self.faker.random_number(3) for _ in range(self.faker.random_number(1))}
                    versions.add(self.faker.random_number(3))
                    for v in versions:
                        content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                        schema_path = self.faker.word()

                        file = os.path.join(base_path, 'endpoints', schema_path) + '.v{}.json'.format(v)
                        if not os.path.isfile(file):
                            patcher.fs.CreateFile(file, contents=json.dumps(content))

                        base_content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                        base_content['$ref'] = 'endpoint://' + schema_path + 'v{}'.format(v)
                        base_contents.append(base_content)

                schema = ISchema()
                subschema = mock.MagicMock()
                subschema.flatten = mock.MagicMock(return_value=False)
                schema._schema_factory = mock.MagicMock(return_value=subschema)
                session = mock.MagicMock()
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                out = yield schema._recurse_subschemas(base_contents, session)
                self.assertEqual(schema.cached, {})
                self.assertEqual(out, {
                    'schema': base_contents,
                    'success': False
                })

    @test_chainable
    def test_recurse_subschemas_endpoint_fails2(self):

        for i in range(5):
            base_path = self.faker.file_path()
            base_content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')

            schema = ISchema()
            old = schema._recurse_subschemas
            schema._recurse_subschemas = mock.MagicMock(return_value={'schema': [], 'success': False})
            session = mock.MagicMock()
            session.component_schemas_path = mock.MagicMock(return_value=base_path)
            out = yield old(base_content, session)
            self.assertEqual(schema.cached, {})
            self.assertEqual(out, {
                'schema': base_content,
                'success': False
            })

    @test_chainable
    def test_recurse_subschemas_endpoint_no_versions(self):

        for i in range(50):
            with Patcher() as patcher:
                base_path = self.faker.file_path()
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                schema_path = self.faker.word()
                content['$ref'] = 'endpoint://' + schema_path

                patcher.fs.CreateFile(os.path.join(base_path, 'endpoints', schema_path) + '.json', contents=json.dumps(content))

                schema = ISchema()
                session = mock.MagicMock()
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                yield self.assertFailure(schema._recurse_subschemas(content, session), RegisterException)

    def test_schema_factory_resource(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            schema = self.faker.word()
            version = self.faker.random_number(1)
            resource = ISchema._schema_factory('resource', '{}/{}/{}/v{}'.format(vendor, component, schema, version))
            self.assertIsInstance(resource, ResourceSchema)
            self.assertEqual(resource.vendor, vendor)
            self.assertEqual(resource.component, component)
            self.assertEqual(resource.schema_path, schema)
            self.assertEqual(resource.versions, [version])

    def test_schema_factory_endpoint(self):

        for i in range(50):
            schema = self.faker.file_path().replace('.', '/')
            version = self.faker.random_number(1)
            endpoint = ISchema._schema_factory('endpoint', '{}/v{}'.format(schema, version))
            self.assertIsInstance(endpoint, EndpointSchema)
            self.assertEqual(endpoint.schema_path, schema)
            self.assertEqual(endpoint.versions, [version])

    def test_schema_factory_http(self):

        for i in range(50):
            schema = self.faker.domain_name()
            endpoint = ISchema._schema_factory('http', schema)
            self.assertIsInstance(endpoint, HttpsSchema)
            self.assertEqual(endpoint.uri, 'https://{}'.format(schema))

    def test_schema_factory_https(self):

        for i in range(50):
            schema = self.faker.domain_name()
            endpoint = ISchema._schema_factory('https', schema)
            self.assertIsInstance(endpoint, HttpsSchema)
            self.assertEqual(endpoint.uri, 'https://{}'.format(schema))

    def test_schema_factory_faulty(self):

        for i in range(50):
            schema = self.faker.domain_name()
            self.assertRaises(RegisterException, ISchema._schema_factory, 'random', schema)

    def test_to_schema(self):
        for i in range(50):
            with Patcher() as patcher:

                base_path = self.faker.file_path()
                schema_path = self.faker.file_path()
                versions = {self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))}
                contents = {}
                for v in versions:
                    content = self.faker.pydict(4, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                    patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v{}.json'.format(v), contents=json.dumps(content))
                    contents[v] = content

                schema = ISchema()
                schema._retrieve_local(base_path, schema_path, versions=versions)
                self.assertEqual(schema.to_schema(), {
                    'oneOf': list(contents.values())
                })

    def test_to_schema_single(self):
        for i in range(50):
            with Patcher() as patcher:
                base_path = self.faker.file_path()
                schema_path = self.faker.file_path()
                version = self.faker.random_number(3)
                content = self.faker.pydict(4, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v{}.json'.format(version), contents=json.dumps(content))

                schema = ISchema()
                schema._retrieve_local(base_path, schema_path, versions=[version])
                self.assertEqual(schema.to_schema(), content)

    def test_to_schema_failure(self):
        schema = ISchema()
        self.assertRaises(NotImplementedError, schema.to_schema)

class InlineSchemaTests(DBTestCase):
    faker = Faker()

    def test_construction(self):
        content = self.faker.pydict(4, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
        schema = InlineSchema(content)
        self.assertEqual(schema.schema, content)
        self.assertEqual(schema.cached, {})

    def test_flatten(self):
        content = self.faker.pydict(4, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
        schema = InlineSchema(content)
        schema._recurse_subschemas = mock.MagicMock(return_value={'test': 'test return'})
        out = schema.flatten({'session': 'test session'})
        self.assertEqual(out, {'test': 'test return'})
        schema._recurse_subschemas.assert_called_once_with(content, {'session': 'test session'})

    def test_to_schema(self):
        content = self.faker.pydict(4, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
        schema = InlineSchema(content)
        self.assertEqual(schema.to_schema(), content)

class HttpsSchemaTests(DBTestCase):
    faker = Faker()

    def test_construction(self):
        uri = self.faker.domain_name()
        schema = HttpsSchema(uri)
        self.assertEqual(schema.uri, 'https://' + uri)
        self.assertEqual(schema.cached, {})

    def test_flatten(self):
        uri = self.faker.domain_name()
        schema = HttpsSchema(uri)
        self.assertTrue(schema.flatten())

    def test_to_schema(self):
        uri = self.faker.domain_name()
        schema = HttpsSchema(uri)
        self.assertEqual(schema.to_schema(), {'$ref': 'https://' + uri})

class EndpointSchemaTests(DBTestCase):
    faker = Faker()

    def test_construction(self):

        for i in range(50):
            path = self.faker.word()
            version = self.faker.random_number(1)
            schema = EndpointSchema('{}/v{}'.format(path, version))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, [version])
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction2(self):

        for i in range(50):
            path = self.faker.word() + '/' + self.faker.word()
            version = self.faker.random_number(1)
            schema = EndpointSchema('{}/v{}'.format(path, version))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, [version])
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction3(self):

        for i in range(50):
            path = self.faker.file_path().replace('.', '/')
            version = self.faker.random_number(1)
            schema = EndpointSchema('{}/v{}'.format(path, version))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, [version])
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction4(self):

        for i in range(50):
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = EndpointSchema('{}/v{}'.format(path, ','.join(str(x) for x in versions)))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions)
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction5(self):

        for i in range(50):
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = EndpointSchema('{}/{}'.format(path, ','.join('v' + str(x) for x in versions)))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions)
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction6(self):

        for i in range(50):
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = EndpointSchema('{}/{}'.format(path, ','.join(str(x) for x in versions)))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions)
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction7(self):

        for i in range(50):
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            versions2 = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = EndpointSchema('{}/v{}'.format(path, ','.join(str(x) for x in versions)), versions=versions2)
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions2)
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction8(self):

        for i in range(50):
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            versions2 = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = EndpointSchema('{}/{}'.format(path, ','.join('v' + str(x) for x in versions)), versions=versions2)
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions2)
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction9(self):

        for i in range(50):
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            versions2 = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = EndpointSchema('{}/{}'.format(path, ','.join(str(x) for x in versions)), versions=versions2)
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions2)
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction10(self):

        for i in range(50):
            path = self.faker.file_path().replace('.', '/')
            schema = EndpointSchema('{}'.format(path))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, [1])
            self.assertEqual(schema.schema_subdir, 'endpoints')
            self.assertEqual(schema.cached, {})

    def test_construction_fail(self):

        for i in range(50):
            path = self.faker.domain_name()
            version = self.faker.random_number(1)
            self.assertRaises(RegisterException, EndpointSchema, '{}/v{}'.format(path, version))

    def test_construction_fail2(self):

        for i in range(50):
            path = self.faker.domain_name()
            version = self.faker.random_number(1)
            self.assertRaises(RegisterException, EndpointSchema, '{}/v-{}'.format(path, version))

    @test_chainable
    def test_flatten(self):

        for i in range(50):
            with Patcher() as patcher:
                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v1.json', contents=json.dumps(content))

                session = mock.MagicMock()
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                schema = EndpointSchema('{}'.format(schema_path))
                self.assertTrue((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {
                    1: content
                })

    @test_chainable
    def test_flatten_cache(self):

        for i in range(50):
            with Patcher() as patcher:
                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v1.json', contents=json.dumps(content))

                session = mock.MagicMock()
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                schema = EndpointSchema('{}'.format(schema_path))
                self.assertTrue((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {
                    1: content
                })
                session.component_schemas_path = mock.MagicMock(return_value=NotImplementedError())
                self.assertTrue((yield schema.flatten(session)))

    @test_chainable
    def test_flatten_failure(self):

        for i in range(50):
            with Patcher() as patcher:
                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v1.json', contents=json.dumps(content))

                session = mock.MagicMock()
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                schema = EndpointSchema('{}'.format(schema_path))
                schema._recurse_subschemas = mock.MagicMock(return_value={'success': False, 'schema': content})
                self.assertFalse((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {})
                self.assertFalse((yield schema.flatten(session)))

class ClaimSchemaTests(DBTestCase):
    faker = Faker()

    def test_construction(self):
        path = self.faker.word()
        version = self.faker.random_number(1)
        schema = ClaimSchema('{}/v{}'.format(path, version))
        self.assertIsInstance(schema, EndpointSchema)
        self.assertEqual(schema.schema_path, path)
        self.assertEqual(schema.versions, [version])
        self.assertEqual(schema.schema_subdir, 'claims')
        self.assertEqual(schema.cached, {})

class ResourceSchemaTests(DBTestCase):
    faker = Faker()

    def test_construction(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.word()
            version = self.faker.random_number(1)
            schema = ResourceSchema('{}/{}/{}/v{}'.format(vendor, component, path, version))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, [version])
            self.assertEqual(schema.cached, {})

    def test_construction2(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.word() + '/' + self.faker.word()
            version = self.faker.random_number(1)
            schema = ResourceSchema('{}/{}/{}/v{}'.format(vendor, component, path, version))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, [version])
            self.assertEqual(schema.cached, {})

    def test_construction3(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.file_path().replace('.', '/')
            version = self.faker.random_number(1)
            schema = ResourceSchema('{}/{}/{}/v{}'.format(vendor, component, path, version))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, [version])
            self.assertEqual(schema.cached, {})

    def test_construction4(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = ResourceSchema('{}/{}/{}/v{}'.format(vendor, component, path, ','.join(str(x) for x in versions)))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions)
            self.assertEqual(schema.cached, {})

    def test_construction5(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = ResourceSchema('{}/{}/{}/{}'.format(vendor, component, path, ','.join('v' + str(x) for x in versions)))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions)
            self.assertEqual(schema.cached, {})

    def test_construction6(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = ResourceSchema('{}/{}/{}/{}'.format(vendor, component, path, ','.join(str(x) for x in versions)))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions)
            self.assertEqual(schema.cached, {})

    def test_construction7(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            versions2 = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = ResourceSchema('{}/{}/{}/v{}'.format(vendor, component, path, ','.join(str(x) for x in versions)), versions=versions2)
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions2)
            self.assertEqual(schema.cached, {})

    def test_construction8(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            versions2 = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = ResourceSchema('{}/{}/{}/{}'.format(vendor, component, path, ','.join('v' + str(x) for x in versions)),
                                    versions=versions2)
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions2)
            self.assertEqual(schema.cached, {})

    def test_construction9(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.file_path().replace('.', '/')
            versions = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            versions2 = list({self.faker.random_number(3) for _ in range(2 + self.faker.random_number(1))})
            schema = ResourceSchema('{}/{}/{}/{}'.format(vendor, component, path, ','.join(str(x) for x in versions)), versions=versions2)
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, versions2)
            self.assertEqual(schema.cached, {})

    def test_construction10(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.file_path().replace('.', '/')
            schema = ResourceSchema('{}/{}/{}'.format(vendor, component, path))
            self.assertEqual(schema.schema_path, path)
            self.assertEqual(schema.versions, [1])
            self.assertEqual(schema.cached, {})

    def test_construction_fail(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.domain_name()
            version = self.faker.random_number(1)
            self.assertRaises(RegisterException, ResourceSchema, '{}/{}/{}/v{}'.format(vendor, component, path, version))

    def test_construction_fail2(self):

        for i in range(50):
            vendor = self.faker.word()
            component = self.faker.word()
            path = self.faker.domain_name()
            version = self.faker.random_number(1)
            self.assertRaises(RegisterException, ResourceSchema, '{}/{}/{}/v-{}'.format(vendor, component, path, version))

    @test_chainable
    def test_flatten(self):

        for i in range(50):
            with Patcher() as _:
                vendor = self.faker.word()
                component = self.faker.word()

                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')

                session = mock.MagicMock()
                group_context = mock.MagicMock()
                group_context.call = mock.MagicMock(return_value=content)
                session.group_context = mock.MagicMock(return_value=group_context)
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                schema = ResourceSchema('{}/{}/{}'.format(vendor, component, schema_path))
                self.assertTrue((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {
                    1: content
                })

                group_context.call.assert_called_once_with('mdstudio.schema.endpoint.get', {
                    'name': schema_path,
                    'version': 1,
                    'component': component,
                    'type': 'resource'
                }, claims={
                    'vendor': vendor
                })

    @test_chainable
    def test_flatten2(self):

        for i in range(50):
            with Patcher() as patcher:
                vendor = self.faker.word()
                component = self.faker.word()

                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v1.json', contents=json.dumps(content))

                session = mock.MagicMock()
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                session.component_config = mock.MagicMock()
                session.component_config.static = mock.MagicMock()
                session.component_config.static.vendor = vendor
                session.component_config.static.component = component

                schema = ResourceSchema('{}/{}/{}'.format(vendor, component, schema_path))
                self.assertTrue((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {
                    1: content
                })

    @test_chainable
    def test_flatten3(self):

        for i in range(50):
            with Patcher() as patcher:
                vendor = self.faker.word()
                component = self.faker.word()

                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v1.json', contents=json.dumps(content))

                session = mock.MagicMock()
                group_context = mock.MagicMock()
                group_context.call = mock.MagicMock(return_value=content)
                session.group_context = mock.MagicMock(return_value=group_context)
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                session.component_config = mock.MagicMock()
                session.component_config.static = mock.MagicMock()
                session.component_config.static.vendor = vendor

                schema = ResourceSchema('{}/{}/{}'.format(vendor, component, schema_path))
                self.assertTrue((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {
                    1: content
                })

                group_context.call.assert_called_once_with('mdstudio.schema.endpoint.get', {
                    'name': schema_path,
                    'version': 1,
                    'component': component,
                    'type': 'resource'
                }, claims={
                    'vendor': vendor
                })

    @test_chainable
    def test_flatten4(self):

        for i in range(50):
            with Patcher() as patcher:
                vendor = self.faker.word()
                component = self.faker.word()

                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v1.json', contents=json.dumps(content))

                session = mock.MagicMock()
                group_context = mock.MagicMock()
                group_context.call = mock.MagicMock(return_value=content)
                session.group_context = mock.MagicMock(return_value=group_context)
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                session.component_config = mock.MagicMock()
                session.component_config.static = mock.MagicMock()
                session.component_config.static.component = component

                schema = ResourceSchema('{}/{}/{}'.format(vendor, component, schema_path))
                self.assertTrue((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {
                    1: content
                })

                group_context.call.assert_called_once_with('mdstudio.schema.endpoint.get', {
                    'name': schema_path,
                    'version': 1,
                    'component': component,
                    'type': 'resource'
                }, claims={
                    'vendor': vendor
                })

    @test_chainable
    def test_flatten_cache(self):

        for i in range(50):
            with Patcher() as patcher:
                vendor = self.faker.word()
                component = self.faker.word()

                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v1.json', contents=json.dumps(content))

                session = mock.MagicMock()
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                session.component_config = mock.MagicMock()
                session.component_config.static = mock.MagicMock()
                session.component_config.static.vendor = vendor
                session.component_config.static.component = component
                schema = ResourceSchema('{}/{}/{}'.format(vendor, component, schema_path))
                self.assertTrue((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {
                    1: content
                })
                session.component_schemas_path = mock.MagicMock(return_value=NotImplementedError())
                self.assertTrue((yield schema.flatten(session)))

    @test_chainable
    def test_flatten_cache2(self):

        for i in range(50):
            with Patcher() as patcher:
                vendor = self.faker.word()
                component = self.faker.word()

                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v1.json', contents=json.dumps(content))

                session = mock.MagicMock()
                group_context = mock.MagicMock()
                group_context.call = mock.MagicMock(return_value=content)
                session.group_context = mock.MagicMock(return_value=group_context)
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                schema = ResourceSchema('{}/{}/{}'.format(vendor, component, schema_path))
                self.assertTrue((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {
                    1: content
                })
                session.component_schemas_path = mock.MagicMock(return_value=NotImplementedError())
                self.assertTrue((yield schema.flatten(session)))

    @test_chainable
    def test_flatten_failure(self):

        for i in range(50):
            with Patcher() as patcher:
                vendor = self.faker.word()
                component = self.faker.word()

                base_path = self.faker.file_path()
                schema_path = self.faker.file_path().replace('.', '/')
                content = self.faker.pydict(10, True, 'str', 'str', 'str', 'str', 'float', 'int', 'int', 'uri', 'email')
                patcher.fs.CreateFile(os.path.join(base_path, schema_path) + '.v1.json', contents=json.dumps(content))

                session = mock.MagicMock()
                session.component_schemas_path = mock.MagicMock(return_value=base_path)
                schema = ResourceSchema('{}/{}/{}'.format(vendor, component, schema_path))
                schema._recurse_subschemas = mock.MagicMock(return_value={'success': False, 'schema': content})
                self.assertFalse((yield schema.flatten(session)))
                self.assertEqual(schema.cached, {})
                self.assertFalse((yield schema.flatten(session)))

class ValidateTests(TestCase):
    def test_validate_json_schema(self):
        validate_json_schema({'format': 'uri', 'type': 'string'}, 'https://example.com')

    def test_validate_json_schema_default(self):
        doc = {'test': 2}
        validate_json_schema({
            'type': 'object',
            'properties': {
                'test': {
                    'type': 'integer'
                },
                'test2': {
                    'type': 'integer',
                    'default': 5
                }
            }
        }, doc)
        self.assertEqual(doc, {
            'test': 2,
            'test2': 5
        })

    def test_validate_json_schema_default2(self):
        doc = {'test': "string"}
        self.assertRaises(ValidationError, validate_json_schema, {
            'type': 'object',
            'properties': {
                'test': {
                    'type': 'integer',
                    'default': 5
                },
                'test2': {
                    'type': 'integer',
                    'default': 5
                }
            }
        }, doc)

    def test_validate_json_schema_fail(self):
        self.assertRaises(ValidationError, validate_json_schema, {'format': 'uri', 'type': 'string'}, 2)

    def test_validate_json_schema_fail2(self):
        self.assertRaises(ValidationError, validate_json_schema, {'format': 'uri', 'type': 'string'}, 'example')

class MDStudioClaimSchemaTests(TestCase):
    def test_construction(self):
        with Patcher() as patcher:
            session = mock.MagicMock()
            session.mdstudio_schemas_path = mock.MagicMock(return_value='schemas')

            patcher.fs.CreateFile('schemas/claims.v1.json', contents='{"test": "json"}')
            schema = MDStudioClaimSchema(session)
            self.assertIs(schema, MDStudioClaimSchema(session))
            self.assertEqual(schema.to_schema(), {'test': 'json'})
            MDStudioClaimSchema._instance = None


text:
<?php

/**
 * Create spot color
 *
 * @phpstub
 *
 * @param resource $psdoc
 * @param string $name
 * @param int $reserved
 *
 * @return int The id of the new spot color or 0 in case of an error.
 */
function ps_makespotcolor($psdoc, $name, $reserved = false)
{
}


text:
<?php

namespace SportChecked\FeatureBundle;

use Symfony\Component\HttpKernel\Bundle\Bundle;

class FeatureBundle extends Bundle
{
}