Dummy - Code Metrics - OnroerendErfgoed/oe_utils - Measure and Improve Code Quality continuously with Scrutinizer

Dummy A
last analyzed 2018-07-04 09:35 UTC

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	4
Duplicated Lines	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	4
rs	10
wmc	0

# -*- coding: utf-8 -*-
import os

import colander
import responses
from sqlalchemy import Column, Integer, Sequence, String, ForeignKey
from sqlalchemy.dialects.postgresql import JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import sessionmaker, relationship
import json

Base = declarative_base()

testdata = {
    "onderwerp": "blabla, boemboem",
    "poststukken": [{
        "id": "poststuk_uri",
        "verwerkt": "geeft aan of een bepaald poststuk reeds verwerkt werd binnen een dossier"
    }],
    "afbeeldingen": ["beeldbank_uri"],
    "documenten": ["document_uri"],
    "procestype": "proces_uri",
    "contactmomenten": [{
        "datum": "",
        "locatie": "crab adres of capakey",
        "korte_inhoud": "",
        "afbeeldingen": ["beeldbank_uri"],
        "documenten": ["document_uri"],
        "externen": ["actor_uri"],
        "actoren": ["actor_uri"],
        "type": "vergadering, plaatsbezoek of mondelinge communicatie"
    }],

    "aangemaakt": "22-03-2011",
    "state": {
        "id": "test",
        "state_qualifier": ""
    },
    "owners": [{
        "type": "actor of groep",
        "id": "owner_actor_uri"
    }],
    "access": [{
        "type": "actor of groep",
        "id": "test_actor_uri"
    }],
    "$schema": "http://localhost/dossierdata/common/0.1.0/",
    "workflow": {
        "id": "",
        "versie": ""
    }
}

incorrect_testdata = "{'555' None}"

dummy_index = {
    "settings": {
        "analysis": {
            "analyzer": {
                "string_lowercase": {
                    "tokenizer": "keyword",
                    "filter": ["lowercase"]
                }
            }
        }
    }
}

dummy_mapping = {
    "properties": {
        "id": {"type": "integer"},
        "onderwerp": {"type": "string"},
        "poststukken": {
            "properties": {
                "id": {"type": "string", "index": "not_analyzed"},
                "relatietype": {"type": "string", "index": "not_analyzed"},
                "verwerkt": {"type": "string"}
            }
        },
        "feed_entry_id": {"type": "string"},
        "afbeeldingen": {
            "properties": {
                "id": {"type": "string", "index": "not_analyzed"},
                "relatietype": {"type": "string", "index": "not_analyzed"}
            }
        },
        "documenten": {
            "properties": {
                "id": {"type": "string", "index": "not_analyzed"},
                "relatietype": {"type": "string", "index": "not_analyzed"}
            }
        },
        "erfgoedobjecten": {
            "properties": {
                "id": {"type": "string", "index": "not_analyzed"},
                "relatietype": {"type": "string", "index": "not_analyzed"}
            }
        },
        "dossiers": {
            "properties": {
                "id": {"type": "string", "index": "not_analyzed"},
                "relatietype": {"type": "string", "index": "not_analyzed"}
            }
        },
        "procestype": {"type": "string"},
        "contactmomenten": {
            "properties": {
                "datum": {"type": "string"},
                "locatie": {"type": "string"},
                "korte_inhoud": {"type": "string"},
                "afbeeldingen": {"type": "string"},
                "documenten": {"type": "string"},
                "externen": {"type": "string"},
                "actoren": {"type": "string"},
                "type": {"type": "string"}
            }
        },
        "aangemaakt": {"type": "string"},
        "state": {
            "properties": {
                "id": {"type": "string"},
                "state_qualifier": {"type": "string"}
            }
        },
        "owners": {
            "type": "nested",
            "properties": {
                "type": {"type": "string"},
                "id": {"type": "string", "index": "not_analyzed"}
            }
        },
        "access": {
            "type": "nested",
            "properties": {
                "type": {"type": "string"},
                "id": {"type": "string", "index": "not_analyzed"}
            }
        },
        "acls": {"type": "string", "index": "not_analyzed"},
        "schema": {"type": "string"},
        "workflow": {"type": "string"},
        "geometry": {
            "type": "geo_shape",
            "properties": {
                "type": {"type": "string"},
                "coordinates": {"type": "multipolygon"},
                "crs": {
                    "properties": {
                        "type": {"type": "string"},
                        "properties": {
                            "properties": {
                                "name": {"type": "string"}
                            }
                        }
                    }
                }
            }
        }
    }
}


def assert_no_exception_raised(callable_function, *args, **kw):
    try:
        callable_function(*args, **kw)
    except Exception as e:
        raise AssertionError('An error was raised: ' + repr(e))


def load_fixture(location):
    with open(os.path.join(os.path.dirname(__file__), location), 'rb') as f:
        data = f.read()
    return data


class DummyDossier(Base):
    '''
    A database representation of a dossier.
    '''
    __tablename__ = 'dossier'
    id = Column(Integer(), Sequence('dossier_id_seq'), primary_key=True)
    feed_entry_id = Column(Integer())
    dossierdata = Column(MutableDict.as_mutable(JSON()))
    version_hash = Column(String())

    emails = relationship(
        'DossierEmail',
        cascade="all, delete, delete-orphan"
    )


class DossierEmail(Base):
    __tablename__ = 'dossieremail'
    dossier_id = Column(Integer, ForeignKey('dossier.id'), primary_key=True)
    email = Column(String(50), primary_key=True)


class Dummy(Base):
    __tablename__ = 'dummy'
    id = Column(Integer(), primary_key=True)
    test = Column(String())


class DummySchema(colander.MappingSchema):
    '''
    A dummy schema to map json data from input and output of requests.
    '''
    test = colander.SchemaNode(colander.String())


def init_test_db(engine):
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    session_maker = sessionmaker(
        bind=engine
    )
    session = session_maker()

    dossier1 = DummyDossier()
    dossier1.feed_entry_id = 666
    dossierdata1 = testdata
    dossierdata1['dossier_id'] = 1
    dossier1.dossierdata = dossierdata1
    dossier1.version_hash = 'U5s/UtvXTczCXIh/EgnFqQ'

    dossier2 = DummyDossier()
    dossier2.feed_entry_id = 123
    dossierdata2 = testdata
    dossierdata2['dossier_id'] = 2
    dossier2.dossierdata = dossierdata2
    dossier2.version_hash = 'juoNihM4QnOU/1pW640n4w'

    session.add(dossier1)
    session.add(dossier2)
    session.add(Dummy(id=1, test='blabla'))
    session.add(Dummy(id=2, test='bla'))
    session.add(Dummy(id=3, test='alpha'))
    session.add(Dummy(id=4, test='zero'))

    session.commit()
    session.close()


def fill_db(session):
    for did in range(75):
        dossier = DummyDossier()
        dossier.feed_entry_id = 666
        dossierdata = testdata
        dossierdata['dossier_id'] = did
        dossier.dossierdata = dossierdata
        dossier.version_hash = 'U5s/UtvXTczCXIh/EgnFqQ'
        session.add(dossier)


def init_search_responses():
    responses.add(responses.PUT, 'http://localhost:9200/dossiers/_mapping/dossier',
                  body=load_fixture('fixtures/response.json'),
                  status=200,
                  content_type='application/json')
    responses.add(responses.PUT, 'http://localhost:9200/dossiers/_mapping/test',
                  body='invalid response',
                  status=409,
                  content_type='application/json')
    responses.add(responses.PUT, 'http://localhost:9200/dossiers',
                  body=load_fixture('fixtures/response.json'),
                  status=200,
                  content_type='application/json')
    responses.add(responses.PUT, 'http://localhost:9200/test',
                  body='invalid response',
                  status=400,
                  content_type='application/json')
    responses.add(responses.HEAD, 'http://localhost:9200/dossiers',
                  body=load_fixture('fixtures/response.json'),
                  status=200,
                  content_type='application/json')
    responses.add(responses.HEAD, 'http://localhost:9200/testing',
                  body='invalid response',
                  status=400,
                  content_type='application/json')
    responses.add(responses.HEAD, 'http://localhost:9200/test',
                  body=load_fixture('fixtures/response.json'),
                  status=200,
                  content_type='application/json')
    responses.add(responses.DELETE, 'http://localhost:9200/dossiers',
                  body=load_fixture('fixtures/response.json'),
                  status=200,
                  content_type='application/json')
    responses.add(responses.DELETE, 'http://localhost:9200/test',
                  body='invalid response',
                  status=400,
                  content_type='application/json')
    responses.add(responses.PUT, 'http://localhost:9200/dossiers/dossier/test1',
                  body=load_fixture('fixtures/response.json'),
                  status=201,
                  content_type='application/json')
    responses.add(responses.PUT, 'http://localhost:9200/dossiers/dossier/test2',
                  body='invalid response',
                  status=400,
                  content_type='application/json')
    responses.add(responses.DELETE, 'http://localhost:9200/dossiers/dossier/test1',
                  body=load_fixture('fixtures/response.json'),
                  status=200,
                  content_type='application/json')
    responses.add(responses.DELETE, 'http://localhost:9200/dossiers/dossier/test2',
                  body='invalid response',
                  status=400,
                  content_type='application/json')
    responses.add(responses.POST, 'http://localhost:9200/dossiers/dossier/_search',
                  body=load_fixture('fixtures/test1.json'),
                  status=200,
                  content_type='application/json')
    responses.add(responses.POST, 'http://localhost:9200/dossiers/testing/_search',
                  body=load_fixture('fixtures/response.json'),
                  status=200,
                  content_type='application/json')
    responses.add(responses.POST, 'http://localhost:9200/dossiers/test/_search',
                  body='invalid response',
                  status=400,
                  content_type='application/json')
    responses.add(responses.POST, 'http://localhost:9200/dossiers/_search',
                  body=load_fixture('fixtures/test1.json'),
                  status=200,
                  content_type='application/json')
    responses.add(responses.POST, 'http://localhost:9200/dossiers/_delete_by_query',
                  status=200,
                  content_type='application/json')



here = os.path.dirname(os.path.abspath(__file__))
with open(os.path.join(here, 'fixtures', 'Test_document.pdf'), 'rb') as test_document:
    data = test_document.read()


def init_document_responses():
    responses.add(responses.POST, 'https://dgen.onroerenderfgoed.be/generations',
                  body=json.dumps({'id': 1}),
                  status=202)
    responses.add(responses.GET, 'https://dgen.onroerenderfgoed.be/generations/1/document',
                  body=data,
                  status=200,
                  content_type='application/pdf')


def init_validation_responses():
    responses.add(responses.GET, "https://id.erfgoed.net/inventaris/teksten/1", status=200,
                  content_type='application/json')


1			# -- coding: utf-8 --
2			import os
3
4			import colander
5			import responses
6			from sqlalchemy import Column, Integer, Sequence, String, ForeignKey
7			from sqlalchemy.dialects.postgresql import JSON
8			from sqlalchemy.ext.declarative import declarative_base
9			from sqlalchemy.ext.mutable import MutableDict
10			from sqlalchemy.orm import sessionmaker, relationship
11			import json
12
13			Base = declarative_base()
14
15			testdata = {
16			"onderwerp": "blabla, boemboem",
17			"poststukken": [{
18			"id": "poststuk_uri",
19			"verwerkt": "geeft aan of een bepaald poststuk reeds verwerkt werd binnen een dossier"
20			}],
21			"afbeeldingen": ["beeldbank_uri"],
22			"documenten": ["document_uri"],
23			"procestype": "proces_uri",
24			"contactmomenten": [{
25			"datum": "",
26			"locatie": "crab adres of capakey",
27			"korte_inhoud": "",
28			"afbeeldingen": ["beeldbank_uri"],
29			"documenten": ["document_uri"],
30			"externen": ["actor_uri"],
31			"actoren": ["actor_uri"],
32			"type": "vergadering, plaatsbezoek of mondelinge communicatie"
33			}],
34
35			"aangemaakt": "22-03-2011",
36			"state": {
37			"id": "test",
38			"state_qualifier": ""
39			},
40			"owners": [{
41			"type": "actor of groep",
42			"id": "owner_actor_uri"
43			}],
44			"access": [{
45			"type": "actor of groep",
46			"id": "test_actor_uri"
47			}],
48			"$schema": "http://localhost/dossierdata/common/0.1.0/",
49			"workflow": {
50			"id": "",
51			"versie": ""
52			}
53			}
54
55			incorrect_testdata = "{'555' None}"
56
57			dummy_index = {
58			"settings": {
59			"analysis": {
60			"analyzer": {
61			"string_lowercase": {
62			"tokenizer": "keyword",
63			"filter": ["lowercase"]
64			}
65			}
66			}
67			}
68			}
69
70			dummy_mapping = {
71			"properties": {
72			"id": {"type": "integer"},
73			"onderwerp": {"type": "string"},
74			"poststukken": {
75			"properties": {
76			"id": {"type": "string", "index": "not_analyzed"},
77			"relatietype": {"type": "string", "index": "not_analyzed"},
78			"verwerkt": {"type": "string"}
79			}
80			},
81			"feed_entry_id": {"type": "string"},
82			"afbeeldingen": {
83			"properties": {
84			"id": {"type": "string", "index": "not_analyzed"},
85			"relatietype": {"type": "string", "index": "not_analyzed"}
86			}
87			},
88			"documenten": {
89			"properties": {
90			"id": {"type": "string", "index": "not_analyzed"},
91			"relatietype": {"type": "string", "index": "not_analyzed"}
92			}
93			},
94			"erfgoedobjecten": {
95			"properties": {
96			"id": {"type": "string", "index": "not_analyzed"},
97			"relatietype": {"type": "string", "index": "not_analyzed"}
98			}
99			},
100			"dossiers": {
101			"properties": {
102			"id": {"type": "string", "index": "not_analyzed"},
103			"relatietype": {"type": "string", "index": "not_analyzed"}
104			}
105			},
106			"procestype": {"type": "string"},
107			"contactmomenten": {
108			"properties": {
109			"datum": {"type": "string"},
110			"locatie": {"type": "string"},
111			"korte_inhoud": {"type": "string"},
112			"afbeeldingen": {"type": "string"},
113			"documenten": {"type": "string"},
114			"externen": {"type": "string"},
115			"actoren": {"type": "string"},
116			"type": {"type": "string"}
117			}
118			},
119			"aangemaakt": {"type": "string"},
120			"state": {
121			"properties": {
122			"id": {"type": "string"},
123			"state_qualifier": {"type": "string"}
124			}
125			},
126			"owners": {
127			"type": "nested",
128			"properties": {
129			"type": {"type": "string"},
130			"id": {"type": "string", "index": "not_analyzed"}
131			}
132			},
133			"access": {
134			"type": "nested",
135			"properties": {
136			"type": {"type": "string"},
137			"id": {"type": "string", "index": "not_analyzed"}
138			}
139			},
140			"acls": {"type": "string", "index": "not_analyzed"},
141			"schema": {"type": "string"},
142			"workflow": {"type": "string"},
143			"geometry": {
144			"type": "geo_shape",
145			"properties": {
146			"type": {"type": "string"},
147			"coordinates": {"type": "multipolygon"},
148			"crs": {
149			"properties": {
150			"type": {"type": "string"},
151			"properties": {
152			"properties": {
153			"name": {"type": "string"}
154			}
155			}
156			}
157			}
158			}
159			}
160			}
161			}
162
163
164			def assert_no_exception_raised(callable_function, args, *kw):
165			try:
166			callable_function(args, *kw)
167			except Exception as e:
168			raise AssertionError('An error was raised: ' + repr(e))
169
170
171			def load_fixture(location):
172			with open(os.path.join(os.path.dirname(__file__), location), 'rb') as f:
173			data = f.read()
174			return data
175
176
177			class DummyDossier(Base):
178			'''
179			A database representation of a dossier.
180			'''
181			__tablename__ = 'dossier'
182			id = Column(Integer(), Sequence('dossier_id_seq'), primary_key=True)
183			feed_entry_id = Column(Integer())
184			dossierdata = Column(MutableDict.as_mutable(JSON()))
185			version_hash = Column(String())
186
187			emails = relationship(
188			'DossierEmail',
189			cascade="all, delete, delete-orphan"
190			)
191
192
193			class DossierEmail(Base):
194			__tablename__ = 'dossieremail'
195			dossier_id = Column(Integer, ForeignKey('dossier.id'), primary_key=True)
196			email = Column(String(50), primary_key=True)
197
198
199			class Dummy(Base):
200			__tablename__ = 'dummy'
201			id = Column(Integer(), primary_key=True)
202			test = Column(String())
203
204
205			class DummySchema(colander.MappingSchema):
206			'''
207			A dummy schema to map json data from input and output of requests.
208			'''
209			test = colander.SchemaNode(colander.String())
210
211
212			def init_test_db(engine):
213			Base.metadata.drop_all(engine)
214			Base.metadata.create_all(engine)
215			session_maker = sessionmaker(
216			bind=engine
217			)
218			session = session_maker()
219
220			dossier1 = DummyDossier()
221			dossier1.feed_entry_id = 666
222			dossierdata1 = testdata
223			dossierdata1['dossier_id'] = 1
224			dossier1.dossierdata = dossierdata1
225			dossier1.version_hash = 'U5s/UtvXTczCXIh/EgnFqQ'
226
227			dossier2 = DummyDossier()
228			dossier2.feed_entry_id = 123
229			dossierdata2 = testdata
230			dossierdata2['dossier_id'] = 2
231			dossier2.dossierdata = dossierdata2
232			dossier2.version_hash = 'juoNihM4QnOU/1pW640n4w'
233
234			session.add(dossier1)
235			session.add(dossier2)
236			session.add(Dummy(id=1, test='blabla'))
237			session.add(Dummy(id=2, test='bla'))
238			session.add(Dummy(id=3, test='alpha'))
239			session.add(Dummy(id=4, test='zero'))
240
241			session.commit()
242			session.close()
243
244
245			def fill_db(session):
246			for did in range(75):
247			dossier = DummyDossier()
248			dossier.feed_entry_id = 666
249			dossierdata = testdata
250			dossierdata['dossier_id'] = did
251			dossier.dossierdata = dossierdata
252			dossier.version_hash = 'U5s/UtvXTczCXIh/EgnFqQ'
253			session.add(dossier)
254
255
256			def init_search_responses():
257			responses.add(responses.PUT, 'http://localhost:9200/dossiers/_mapping/dossier',
258			body=load_fixture('fixtures/response.json'),
259			status=200,
260			content_type='application/json')
261			responses.add(responses.PUT, 'http://localhost:9200/dossiers/_mapping/test',
262			body='invalid response',
263			status=409,
264			content_type='application/json')
265			responses.add(responses.PUT, 'http://localhost:9200/dossiers',
266			body=load_fixture('fixtures/response.json'),
267			status=200,
268			content_type='application/json')
269			responses.add(responses.PUT, 'http://localhost:9200/test',
270			body='invalid response',
271			status=400,
272			content_type='application/json')
273			responses.add(responses.HEAD, 'http://localhost:9200/dossiers',
274			body=load_fixture('fixtures/response.json'),
275			status=200,
276			content_type='application/json')
277			responses.add(responses.HEAD, 'http://localhost:9200/testing',
278			body='invalid response',
279			status=400,
280			content_type='application/json')
281			responses.add(responses.HEAD, 'http://localhost:9200/test',
282			body=load_fixture('fixtures/response.json'),
283			status=200,
284			content_type='application/json')
285			responses.add(responses.DELETE, 'http://localhost:9200/dossiers',
286			body=load_fixture('fixtures/response.json'),
287			status=200,
288			content_type='application/json')
289			responses.add(responses.DELETE, 'http://localhost:9200/test',
290			body='invalid response',
291			status=400,
292			content_type='application/json')
293			responses.add(responses.PUT, 'http://localhost:9200/dossiers/dossier/test1',
294			body=load_fixture('fixtures/response.json'),
295			status=201,
296			content_type='application/json')
297			responses.add(responses.PUT, 'http://localhost:9200/dossiers/dossier/test2',
298			body='invalid response',
299			status=400,
300			content_type='application/json')
301			responses.add(responses.DELETE, 'http://localhost:9200/dossiers/dossier/test1',
302			body=load_fixture('fixtures/response.json'),
303			status=200,
304			content_type='application/json')
305			responses.add(responses.DELETE, 'http://localhost:9200/dossiers/dossier/test2',
306			body='invalid response',
307			status=400,
308			content_type='application/json')
309			responses.add(responses.POST, 'http://localhost:9200/dossiers/dossier/_search',
310			body=load_fixture('fixtures/test1.json'),
311			status=200,
312			content_type='application/json')
313			responses.add(responses.POST, 'http://localhost:9200/dossiers/testing/_search',
314			body=load_fixture('fixtures/response.json'),
315			status=200,
316			content_type='application/json')
317			responses.add(responses.POST, 'http://localhost:9200/dossiers/test/_search',
318			body='invalid response',
319			status=400,
320			content_type='application/json')
321			responses.add(responses.POST, 'http://localhost:9200/dossiers/_search',
322			body=load_fixture('fixtures/test1.json'),
323			status=200,
324			content_type='application/json')
325			responses.add(responses.POST, 'http://localhost:9200/dossiers/_delete_by_query',
326			status=200,
327			content_type='application/json')
328
329
330
331			here = os.path.dirname(os.path.abspath(__file__))
332			with open(os.path.join(here, 'fixtures', 'Test_document.pdf'), 'rb') as test_document:
333			data = test_document.read()
334
335
336			def init_document_responses():
337			responses.add(responses.POST, 'https://dgen.onroerenderfgoed.be/generations',
338			body=json.dumps({'id': 1}),
339			status=202)
340			responses.add(responses.GET, 'https://dgen.onroerenderfgoed.be/generations/1/document',
341			body=data,
342			status=200,
343			content_type='application/pdf')
344
345
346			def init_validation_responses():
347			responses.add(responses.GET, "https://id.erfgoed.net/inventaris/teksten/1", status=200,
348			content_type='application/json')
349

OnroerendErfgoed / oe_utils

Dummy A last analyzed 2018-07-04 09:35 UTC

Complexity

Size/Duplication

Importance

Duplication Side-by-Side

Filter issues like

Dummy A
last analyzed 2018-07-04 09:35 UTC