Commit 9b877120 authored by Julien David's avatar Julien David
Browse files

Merge branch 'Adding-Json-FUNCTIONS' into 'master'

Adding json functions

See merge request !5
parents fd433984 246939d0
Pipeline #718 passed with stage
in 27 seconds
softwares/server/__pycache__/__init__.cpython-36.pyc
softwares/server/__pycache__/server.cpython-36.pyc
softwares/server/.vscode/settings.json
softwares/database/__init__.py
softwares/database/.vscode/settings.json
tests/test_server.py
softwares/.vscode/settings.json
tests/.vscode/settings.json
.vscode/settings.json
softwares/__init__.py
softwares/__pycache__/__init__.cpython-38.pyc
softwares/server/__pycache__/__init__.cpython-38.pyc
softwares/server/__pycache__/server.cpython-38.pyc
tests/__init__.py
tests/.vscode/settings.json
tests/.vscode/settings.json
tests/__pycache__/test_secret.cpython-38-pytest-6.0.2.pyc
tests/.vscode/settings.json
tests/.vscode/settings.json
tests/__pycache__/__init__.cpython-38.pyc
tests/__pycache__/test_query_valid.cpython-38-pytest-6.0.2.pyc
tests/__pycache__/test_mail.cpython-38-pytest-6.0.2.pyc
softwares/mailer/__init__.py
softwares/server/__init__.py
# Authors : Julien David & Ismail MOUMNI
# authors: Julien DAVID & Ismail MOUMNI
import socket
import json
import os
import sys
import database.database as db
sys.path.append(os.path.realpath('../softwares/'))
# SERVER PORT
__RDOS_Port__ = 9393
# Creating Socket IP4 TCP
def server_conn(address, PORT):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
# BINDING
serv.bind((address, PORT))
# Listening to 1 CLient
serv.listen()
# Accepting the connection from client and getting client IP
conn, addr = serv.accept()
with conn:
# Sending
conn.send(bytes("Bonjour !!", "utf-8"))
print('Connexion acceptée depuis l IP : ', addr)
# Receiving Data from Client
data = conn.recv(1024)
print(data)
serv.close()
__RDOS_Dict__ = {}
# tools List
__RDOS_Tool__ = []
class RdosServer:
# Function server_conn creates a socket and listens on port 9393
# Function input address : SERVER IP ADRESS & Port server Port
# Function output query send from client to server
def server_conn(address, PORT):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
serv.bind((address, PORT))
# Listening to 5 CLients
serv.listen(5)
# accepting the connection from client and getting client IP
conn, addr = serv.accept()
while True:
if conn:
print('Connexion acceptee depuis l IP : ', addr)
# Receiving Data from Client
data = conn.recv(4096)
query = (json.loads(data.decode('utf-8')))
print("Requete reçu : ", query)
match = verif_param(query)
if match != query:
print('Erreur query : ', match)
conn.send(bytes(json.dumps(match), "utf-8"))
else:
print("ok")
conn.send(bytes(json.dumps(match), "utf-8"))
# Function query_valwid matches the query send from client side
# with the the parameter dict
# Function input dictionary from client
# Function output boolean true if match else False
def query_valid(data: dict):
if 'parameters' in data:
if (data['parameters'] != '') & (len(data) <= 2):
return True
else:
return False
# Function json_loads changes a json file into a Dict
# Function json_loads returns a dictionary
def json_to_dict(js):
data = json.loads(js)
return data
# Function db_req sends a request to get parameters of tools from database
# Function takes a dictionary for input
# Function return a Dict containing tool parameters and default parameters
def db_req(tool: dict):
tool_dict = db.getParameters(tool)
return tool_dict
# Function match_dict returns non existing fields in a dict
# Function input biblio and dict
# Function returns a list containing missing values
def missing_keys(biblio: dict, gen: dict):
missing = []
if(biblio is not None and gen is not None):
for key in biblio:
if key not in gen:
missing.append(key)
return missing
else:
raise Exception("Dictionnaire Vide!!")
# Function match_query_dict matches two dictionaries
# Function input takes 2 dictionaries to match
# Function output return trur if dictionaries matches else raise a value error exception
def match_query_dict(biblio: dict, data: dict):
if biblio is not None and data is not None:
return biblio.keys() == data.keys()
else:
raise ValueError("Error Dict Values ")
# Function verif_param verifies json send from client to server by matching it with database query
# Function input takes a dictionary
# Function output returns query if it matches with database query keys if not it raises an error
def verif_param(data: dict):
tool = data.keys().strip('[]')
s = db.getParameters(data['tool'])
parameter = {}
query = {}
if s is not None:
gen = s[0]
gen_default = s[1]
parameter = gen_default
query = data.get(tool)
if match_query_dict(gen.get(tool), query):
for a in query:
if query[a] == "":
print(query[a])
query[a] = parameter.get(a)
return query
else:
raise Exception("Query Doesn't match")
else:
raise Exception("Error Database")
import os
import sys
import socket
sys.path.append(os.path.realpath('../softwares/'))
import mailer.mail_templates as ms
import mailer.mail_templates as ms # noqa E402
# ensures that the connection is timed out after 1 second (in cases the test is not
# launched on the proper network
import socket
socket.setdefaulttimeout(1.0)
socket.setdefaulttimeout(1.0)
def test_empty_mail():
assert ms.send_mail("","") == 0
assert ms.send_mail("","1234") == 0
assert ms.send_mail("", "") == 0
assert ms.send_mail("", "1234") == 0
#assert ms.send_mail("julien.david@lipn.univ-paris13.fr","1234") == -1
# assert ms.send_mail("julien.david@lipn.univ-paris13.fr","1234") == -1
#!/usr/bin/env python3
# Authors : Julien DAVID & Ismail MOUMNI
import sys
import os
import pytest
sys.path.append(os.path.realpath('../softwares/'))
import server.server as srv # noqa E402
data1 = {"id": "", "name": "", "state": "", "launch_date": "", "mdr_number": "", "nameC": {"first": "", "last": ""}, "title": "", "id": ""}
data2 = {"id": "", "name": "", "state": "", "launch_date": "", "mdr_number": "", "id": ""}
def test_match_query():
var = srv.match_query_dict(data1, data1)
assert var is not False
def test_diff_match_query():
result = srv.match_query_dict(data1, data2)
assert result is not None
def test_bad_match():
with pytest.raises(Exception):
assert srv.match_query_dict({}, data2)
#!/usr/bin/env python3
# Authors : Julien DAVID & Ismail MOUMNI
import sys
import os
import json
import pytest
sys.path.append(os.path.realpath('../softwares/'))
import server.server as srv # noqa E402
data1 = '{"id": "", "name": "", "state": "", "launch_date": "", "mdr_number": "", "nameC": {"first": "", "last": ""}, "title": "", "id": ""}'
data2 = '{"id": "", "name": "", "state": "", "launch_date": "", "mdr_number": "", "id": ""}'
obj = (json.loads(data1))
obj2 = (json.loads(data2))
def test_missing_json():
var = srv.missing_keys(obj, obj)
assert var == []
def test_diff_missing_json():
result = srv.missing_keys(obj, obj2)
assert result is not None
def test_bad_match_json():
with pytest.raises(Exception):
assert srv.missing_keys({}, obj)
#!/usr/bin/ python3
# Authors : Julien DAVID & Ismail MOUMNI
import sys
import os
import pytest
sys.path.append(os.path.realpath('../softwares/'))
import server.server as srv # noqa E402
data1 = {
"id": "",
"name": "",
"state": "",
"launch_date": "",
"mdr_number": "",
"nces_id": "",
"error": "",
"sis_type": "",
"pause_start": "",
"pause_end": "",
"district_contact": {
"district": "",
"email": "",
"name": {
"first": "",
"last": "",
},
"title": "",
"id": ""
}
}
data2 = {
"id": "",
"name": "",
"state": "",
"launch_date": "",
"mdr_number": "",
"nces_id": "",
"error": "",
"sis_type": "",
"pause_start": "",
"pause_end": "",
"district_contact": {
"district": "",
"email": "",
"name": {
"first": "",
"last": "",
},
"title": "",
"id": ""
}
}
def test_missing_keys():
var = srv.missing_keys(data1, data2)
assert not var
def test_bad_missing():
with pytest.raises(Exception):
assert srv.missing_keys({}, data1)
#!/usr/bin/ python3
# Authors : Julien DAVID & Ismail MOUMNI
import sys
import os
import pytest
import json
sys.path.append(os.path.realpath('../softwares/'))
import server.server as serv # noqa E402
data_test1 = '{"parameters": "request"}'
data_test2 = '{"Ref": "Parameters"}'
data_load = json.loads((data_test1))
data_load2 = json.loads(data_test2)
def test_query_valid():
var = serv.query_valid(data_load)
assert ((var) is True)
def test_bad_missing():
with pytest.raises(Exception):
assert serv.query_valid(data_load2)
......@@ -3,7 +3,9 @@ import os
import sys
sys.path.append(os.path.realpath('../softwares/'))
import database.database as db
import database.database as db # noqa E402
def test_empty_password():
print("Empty password test")
with pytest.raises(db.NoPasswordDefineError):
......@@ -11,8 +13,6 @@ def test_empty_password():
def test_unsecurred_password():
print("Unsecured password test")
print("Unsecured password test")
os.environ['rdos_secret'] = "test"
assert db.get_password() == "test"
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment