Compare commits

..

21 Commits

Author SHA1 Message Date
9b27b071f7 let only admin 2025-08-28 10:01:37 +02:00
ceb9cd30b7 licence 2025-08-28 09:59:30 +02:00
c000c8fd65 first readme 2025-08-28 09:58:34 +02:00
7b34a37788 clean config variables names 2025-08-28 07:19:35 +02:00
89bb69dce5 init db with test data 2025-08-28 06:39:20 +02:00
55f2be4870 test collaborator`s list on team_id 2025-08-27 11:07:57 +02:00
1eb19b115b remove create test; need inputs 2025-08-27 10:18:17 +02:00
f4d380b3ce install coverage 2025-08-27 10:17:42 +02:00
a0c0bf7931 more collab tests 2025-08-27 10:15:30 +02:00
f492b12479 fixture pytest db sqlite in mem 2025-08-25 17:00:57 +02:00
2b6d6893f3 add sentry 2025-08-25 16:58:41 +02:00
523b1d55d5 submenu to choose customer or user 2025-08-25 16:57:26 +02:00
425565b639 fix test sqlite iso issue on date 2025-08-25 16:21:29 +02:00
50d6e8f5b7 filter objects on connected user 2025-08-25 16:20:40 +02:00
4511cb312b distinct tools tests by classes 2025-08-25 16:17:47 +02:00
20b72f7288 made auth test user/pw simplier 2025-08-22 12:07:09 +02:00
ef2688b61a test dirs + fixture session/init in conftest 2025-08-22 11:53:08 +02:00
eb11a3cd04 add docstring and call view instead prints 2025-08-22 11:52:05 +02:00
c7c76b3bef connection test if user doesn`t exist 2025-08-22 11:50:01 +02:00
01d05e276a move init db into main 2025-08-22 11:48:10 +02:00
34b0ea41e5 adjust filter for no support event 2025-08-22 11:47:41 +02:00
19 changed files with 551 additions and 46 deletions

21
LICENCE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 ylxdre
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

View File

@@ -10,6 +10,7 @@ flake8 = "*"
flake8-html = "*"
pytest = "*"
sentry-sdk = "*"
pytest-cov = "*"
[dev-packages]

114
Pipfile.lock generated
View File

@@ -1,7 +1,7 @@
{
"_meta": {
"hash": {
"sha256": "b3e0227c7afe29a83bdcb59401e9133cb2ee204d2c6dcd0d6e6d00b9781f0558"
"sha256": "3f8aaf05ed2431c2ff68e6d588c0a121fc868f90613a6533a25de7b9a1945dd4"
},
"pipfile-spec": 6,
"requires": {
@@ -24,6 +24,103 @@
"markers": "python_version >= '3.7'",
"version": "==2025.8.3"
},
"coverage": {
"extras": [
"toml"
],
"hashes": [
"sha256:02252dc1216e512a9311f596b3169fad54abcb13827a8d76d5630c798a50a754",
"sha256:02650a11324b80057b8c9c29487020073d5e98a498f1857f37e3f9b6ea1b2426",
"sha256:03f47dc870eec0367fcdd603ca6a01517d2504e83dc18dbfafae37faec66129a",
"sha256:0520dff502da5e09d0d20781df74d8189ab334a1e40d5bafe2efaa4158e2d9e7",
"sha256:0666cf3d2c1626b5a3463fd5b05f5e21f99e6aec40a3192eee4d07a15970b07f",
"sha256:0913dd1613a33b13c4f84aa6e3f4198c1a21ee28ccb4f674985c1f22109f0aae",
"sha256:0be24d35e4db1d23d0db5c0f6a74a962e2ec83c426b5cac09f4234aadef38e4a",
"sha256:0d511dda38595b2b6934c2b730a1fd57a3635c6aa2a04cb74714cdfdd53846f4",
"sha256:146fa1531973d38ab4b689bc764592fe6c2f913e7e80a39e7eeafd11f0ef6db2",
"sha256:14d6071c51ad0f703d6440827eaa46386169b5fdced42631d5a5ac419616046f",
"sha256:1b7181c0feeb06ed8a02da02792f42f829a7b29990fef52eff257fef0885d760",
"sha256:1d043a8a06987cc0c98516e57c4d3fc2c1591364831e9deb59c9e1b4937e8caf",
"sha256:1f64b8d3415d60f24b058b58d859e9512624bdfa57a2d1f8aff93c1ec45c429b",
"sha256:1f672efc0731a6846b157389b6e6d5d5e9e59d1d1a23a5c66a99fd58339914d5",
"sha256:1f8a81b0614642f91c9effd53eec284f965577591f51f547a1cbeb32035b4c2f",
"sha256:2285c04ee8676f7938b02b4936d9b9b672064daab3187c20f73a55f3d70e6b4a",
"sha256:2968647e3ed5a6c019a419264386b013979ff1fb67dd11f5c9886c43d6a31fc2",
"sha256:2b96bfdf7c0ea9faebce088a3ecb2382819da4fbc05c7b80040dbc428df6af44",
"sha256:2d1b73023854068c44b0c554578a4e1ef1b050ed07cf8b431549e624a29a66ee",
"sha256:2d488d7d42b6ded7ea0704884f89dcabd2619505457de8fc9a6011c62106f6e5",
"sha256:32ddaa3b2c509778ed5373b177eb2bf5662405493baeff52278a0b4f9415188b",
"sha256:343a023193f04d46edc46b2616cdbee68c94dd10208ecd3adc56fcc54ef2baa1",
"sha256:36d42b7396b605f774d4372dd9c49bed71cbabce4ae1ccd074d155709dd8f235",
"sha256:384b34482272e960c438703cafe63316dfbea124ac62006a455c8410bf2a2262",
"sha256:3876385722e335d6e991c430302c24251ef9c2a9701b2b390f5473199b1b8ebf",
"sha256:38a9109c4ee8135d5df5505384fc2f20287a47ccbe0b3f04c53c9a1989c2bbaf",
"sha256:3f39cef43d08049e8afc1fde4a5da8510fc6be843f8dea350ee46e2a26b2f54c",
"sha256:4028e7558e268dd8bcf4d9484aad393cafa654c24b4885f6f9474bf53183a82a",
"sha256:414a568cd545f9dc75f0686a0049393de8098414b58ea071e03395505b73d7a8",
"sha256:42144e8e346de44a6f1dbd0a56575dd8ab8dfa7e9007da02ea5b1c30ab33a7db",
"sha256:44d43de99a9d90b20e0163f9770542357f58860a26e24dc1d924643bd6aa7cb4",
"sha256:467dc74bd0a1a7de2bedf8deaf6811f43602cb532bd34d81ffd6038d6d8abe99",
"sha256:5255b3bbcc1d32a4069d6403820ac8e6dbcc1d68cb28a60a1ebf17e47028e898",
"sha256:54a1532c8a642d8cc0bd5a9a51f5a9dcc440294fd06e9dda55e743c5ec1a8f14",
"sha256:556d23d4e6393ca898b2e63a5bca91e9ac2d5fb13299ec286cd69a09a7187fde",
"sha256:5661bf987d91ec756a47c7e5df4fbcb949f39e32f9334ccd3f43233bbb65e508",
"sha256:585ffe93ae5894d1ebdee69fc0b0d4b7c75d8007983692fb300ac98eed146f78",
"sha256:5e78bd9cf65da4c303bf663de0d73bf69f81e878bf72a94e9af67137c69b9fe9",
"sha256:5f1dc8f1980a272ad4a6c84cba7981792344dad33bf5869361576b7aef42733a",
"sha256:6013a37b8a4854c478d3219ee8bc2392dea51602dd0803a12d6f6182a0061762",
"sha256:609b60d123fc2cc63ccee6d17e4676699075db72d14ac3c107cc4976d516f2df",
"sha256:61f78c7c3bc272a410c5ae3fde7792b4ffb4acc03d35a7df73ca8978826bb7ab",
"sha256:62835c1b00c4a4ace24c1a88561a5a59b612fbb83a525d1c70ff5720c97c0610",
"sha256:63d4bb2966d6f5f705a6b0c6784c8969c468dbc4bcf9d9ded8bff1c7e092451f",
"sha256:63df1fdaffa42d914d5c4d293e838937638bf75c794cf20bee12978fc8c4e3bc",
"sha256:66c644cbd7aed8fe266d5917e2c9f65458a51cfe5eeff9c05f15b335f697066e",
"sha256:672a6c1da5aea6c629819a0e1461e89d244f78d7b60c424ecf4f1f2556c041d8",
"sha256:68c5e0bc5f44f68053369fa0d94459c84548a77660a5f2561c5e5f1e3bed7031",
"sha256:6a29f8e0adb7f8c2b95fa2d4566a1d6e6722e0a637634c6563cb1ab844427dd9",
"sha256:6b87f1ad60b30bc3c43c66afa7db6b22a3109902e28c5094957626a0143a001f",
"sha256:73269df37883e02d460bee0cc16be90509faea1e3bd105d77360b512d5bb9c33",
"sha256:74d5b63fe3f5f5d372253a4ef92492c11a4305f3550631beaa432fc9df16fcff",
"sha256:7e78b767da8b5fc5b2faa69bb001edafcd6f3995b42a331c53ef9572c55ceb82",
"sha256:7fa22800f3908df31cea6fb230f20ac49e343515d968cc3a42b30d5c3ebf9b5a",
"sha256:8002dc6a049aac0e81ecec97abfb08c01ef0c1fbf962d0c98da3950ace89b869",
"sha256:8048ce4b149c93447a55d279078c8ae98b08a6951a3c4d2d7e87f4efc7bfe100",
"sha256:90dc3d6fb222b194a5de60af8d190bedeeddcbc7add317e4a3cd333ee6b7c879",
"sha256:9a86281794a393513cf117177fd39c796b3f8e3759bb2764259a2abba5cce54b",
"sha256:a46473129244db42a720439a26984f8c6f834762fc4573616c1f37f13994b357",
"sha256:a931a87e5ddb6b6404e65443b742cb1c14959622777f2a4efd81fba84f5d91ba",
"sha256:ad8fa9d5193bafcf668231294241302b5e683a0518bf1e33a9a0dfb142ec3031",
"sha256:b08801e25e3b4526ef9ced1aa29344131a8f5213c60c03c18fe4c6170ffa2874",
"sha256:b0ef4e66f006ed181df29b59921bd8fc7ed7cd6a9289295cd8b2824b49b570df",
"sha256:b3dcf2ead47fa8be14224ee817dfc1df98043af568fe120a22f81c0eb3c34ad2",
"sha256:b45264dd450a10f9e03237b41a9a24e85cbb1e278e5a32adb1a303f58f0017f3",
"sha256:b4fdc777e05c4940b297bf47bf7eedd56a39a61dc23ba798e4b830d585486ca5",
"sha256:bc85eb2d35e760120540afddd3044a5bf69118a91a296a8b3940dfc4fdcfe1e2",
"sha256:bc8e4d99ce82f1710cc3c125adc30fd1487d3cf6c2cd4994d78d68a47b16989a",
"sha256:c177e6ffe2ebc7c410785307758ee21258aa8e8092b44d09a2da767834f075f2",
"sha256:c2492e4dd9daab63f5f56286f8a04c51323d237631eb98505d87e4c4ff19ec34",
"sha256:c2d05c7e73c60a4cecc7d9b60dbfd603b4ebc0adafaef371445b47d0f805c8a9",
"sha256:c6a5c3414bfc7451b879141ce772c546985163cf553f08e0f135f0699a911801",
"sha256:cebd8e906eb98bb09c10d1feed16096700b1198d482267f8bf0474e63a7b8d84",
"sha256:cf33134ffae93865e32e1e37df043bef15a5e857d8caebc0099d225c579b0fa3",
"sha256:d9cd64aca68f503ed3f1f18c7c9174cbb797baba02ca8ab5112f9d1c0328cd4b",
"sha256:dd382410039fe062097aa0292ab6335a3f1e7af7bba2ef8d27dcda484918f20c",
"sha256:e551f9d03347196271935fd3c0c165f0e8c049220280c1120de0084d65e9c7ff",
"sha256:eb7b0bbf7cc1d0453b843eca7b5fa017874735bef9bfdfa4121373d2cc885ed6",
"sha256:eb90fe20db9c3d930fa2ad7a308207ab5b86bf6a76f54ab6a40be4012d88fcae",
"sha256:ed9749bb8eda35f8b636fb7632f1c62f735a236a5d4edadd8bbcc5ea0542e732",
"sha256:ef3b83594d933020f54cf65ea1f4405d1f4e41a009c46df629dd964fcb6e907c",
"sha256:f2e57716a78bc3ae80b2207be0709a3b2b63b9f2dcf9740ee6ac03588a2015b6",
"sha256:f366a57ac81f5e12797136552f5b7502fa053c861a009b91b80ed51f2ce651c6",
"sha256:f39071caa126f69d63f99b324fb08c7b1da2ec28cbb1fe7b5b1799926492f65c",
"sha256:f4446a9547681533c8fa3e3c6cf62121eeee616e6a92bd9201c6edd91beffe13",
"sha256:f9559b906a100029274448f4c8b8b0a127daa4dade5661dfd821b8c188058842",
"sha256:fcf6ab569436b4a647d4e91accba12509ad9f2554bc93d3aee23cc596e7f99c3",
"sha256:fefafcca09c3ac56372ef64a40f5fe17c5592fab906e0fdffd09543f3012ba50"
],
"markers": "python_version >= '3.9'",
"version": "==7.10.5"
},
"exceptiongroup": {
"hashes": [
"sha256:4d111e6e0c13d0644cad6ddaa7ed0261a0b36971f6d23e7ec9b4b9097da78a10",
@@ -283,6 +380,15 @@
"markers": "python_version >= '3.9'",
"version": "==8.4.1"
},
"pytest-cov": {
"hashes": [
"sha256:25cc6cc0a5358204b8108ecedc51a9b57b34cc6b8c967cc2c01a4e00d8a67da2",
"sha256:f5bc4c23f42f1cdd23c70b1dab1bbaef4fc505ba950d53e0081d0730dd7e86d5"
],
"index": "pypi",
"markers": "python_version >= '3.9'",
"version": "==6.2.1"
},
"sentry-sdk": {
"hashes": [
"sha256:5ea58d352779ce45d17bc2fa71ec7185205295b83a9dbb5707273deb64720092",
@@ -396,11 +502,11 @@
},
"typing-extensions": {
"hashes": [
"sha256:38b39f4aeeab64884ce9f74c94263ef78f3c22467c8724005483154c26648d36",
"sha256:d1e1e3b58374dc93031d6eda2420a48ea44a36c2b4766a4fdeb3710755731d76"
"sha256:0cea48d173cc12fa28ecabc3b837ea3cf6f38c6d1136f85cbaaf598984861466",
"sha256:f0fa19c6845758ab08074a0cfa8b7aecb71c999ca73d62883bc25cc018c4e548"
],
"markers": "python_version >= '3.9'",
"version": "==4.14.1"
"version": "==4.15.0"
},
"urllib3": {
"hashes": [

105
README.md Normal file
View File

@@ -0,0 +1,105 @@
# OCR / DA Python - Project12
## Epic Events
Build a secure backend architecture with Python and SQL
A CLI application allowing three different type of users to connect
(permissions management) and to manage customers, contracts and events
### Introduction
These instructions allow you to :
- get the program
- install the required environment
- run and use it
---
### Requirements
1. modules
```
python 3.10, python3.10-venv, git, pipenv,
SQLAlchemy, mysql-connector-python, passlib, argon2, simple-term-menu, sentry-sdk
```
2. applications
```
mysql-server-8.0, mysql-client
```
You'll need to create a dedicated DB and user to your application
### Installation
1. Clone this repo and go in the project's directory
2. Create the virtual environment and install dependencies
```
pipenv sync
```
3. Enter the venv
```
pipenv shell
```
4. Setup your database
Once your MySQL server is installed and running :
- connect with the root user
```
mysql -u root -p
```
- create a database
```
CREATE DATABASE MYDB
```
- create user and grant permissions
```
CREATE USER 'MYUSER'@'%' IDENTIFIED BY 'MYPASSWORD';
GRANT ALL PRIVILEGES ON MYDB TO 'MYUSER'@'%';
```
5. Create your configuration's file
Create a file named `config.py` in your app's directory containing (at least):
```
USER = your_db_user
PASSWORD = your_db_user_password
```
6. Sentry
This app can use Sentry. You just have to place the server's URL containing your
key in the `config.py` file :
```
SENTRY_URL = your_url
```
---
### Execution
1. Go into the app directory
2. Launch the app
```
python main.py
```
___
### Usage
1. First connection
Connect first with the default 'admin' user and 'password' as password, then create new collaborators following the menu
2. Users
There are three roles having three kinds of permissions. The menu is launched accordingly depending on the type of user logged in.
First level is objects (Collaborators, Customers, Contracts, Event), second level is CRUD depending on permissions, and there are some filter on 'list'.
-----
### Author
YaL <yann@needsome.coffee>
### License
MIT License
Copyright (c) 2025

View File

@@ -39,9 +39,6 @@ class PasswordTools:
def check(self, username: str, password: str) -> bool:
user = self.get_by_name(username)
if not user:
print("Wrong user")
return False
user_pw = user.password_hash
return argon2.verify(password, user_pw)

View File

@@ -33,30 +33,34 @@ class App:
:return: tuple(team_id:int, user_id:int) | None and print if no match
"""
username, password = self.view.prompt_connect()
if not self.collaborator_tools.get_id_by_name(username):
self.view.display_no_user()
quit()
else:
if self.passwd_tools.check(username, password):
perm = self.collaborator_tools.get_team_by_name(username)
user_id = self.collaborator_tools.get_id_by_name(username)
return perm, user_id
else:
print("Connection failed")
return None
self.view.display_co_failed()
quit()
def start(self):
team, user_id = self.connect()
if not user_id:
exit()
if team == 1:
CommercialMenu(self.customer_tools,
self.contract_tools,
self.event_tools,
self.tools).launch()
self.tools,
user_id).launch()
if team == 2:
ManagementMenu(self.collaborator_tools,
self.customer_tools,
self.contract_tools,
self.event_tools,
self.tools).launch()
self.tools,
user_id).launch()
if team == 3:
SupportMenu(self.customer_tools,

5
db.py
View File

@@ -1,9 +1,10 @@
from sqlalchemy import create_engine
from config import user1, pass1
from config import USER, PASSWORD
from sqlalchemy.orm import sessionmaker
DB_URL = "mysql+mysqlconnector://"+user1+":"+pass1+"@localhost:3306/prout"
DB = "epicevents"
DB_URL = "mysql+mysqlconnector://"+USER+":"+PASSWORD+"@localhost:3306/"+DB
engine = create_engine(DB_URL, echo=False)

44
initdb.py Normal file
View File

@@ -0,0 +1,44 @@
from types import resolve_bases
import models
from db import engine
from sqlalchemy import MetaData
from passlib.hash import argon2
def write_db(db, model):
db.add(model)
db.commit()
def clean_db():
models.Base.metadata.drop_all(bind=engine)
def init_test_db(db):
# clean up everything before starting
# models.Attendee.__table__.drop(engine)
teams = ["commercial", "management", "support"]
for item in teams:
team = models.Team(name=item)
write_db(db, team)
# create a manager
man = models.Collaborator(name="admin",
email="a",
phone=1,
team_id=2,
)
write_db(db, man)
man_password = models.Credentials(
collaborator_id = man.id,
password_hash = argon2.hash("password"))
write_db(db, man_password)
# create an attendee
attendee = models.Attendee(name="Guest")
write_db(db, attendee)
# data = [com1, commercial_password, attendee]

13
main.py
View File

@@ -4,12 +4,21 @@ from db import engine, session
from controllers import App
from views import View
from authentication import PasswordTools
from initdb import init_test_db
import sentry_sdk
import config
models.Base.metadata.create_all(bind=engine)
def main():
sentry_sdk.init(
dsn=config.SENTRY_URL,
# Add data like request headers and IP for users,
# see https://docs.sentry.io/platforms/python/data-management/data-collected/ for more info
send_default_pii=True,
)
view = View()
collaborator_tools = tools.CollaboratorTools(session)
customer_tools = tools.CustomerTools(session)
@@ -29,4 +38,6 @@ def main():
if __name__ == "__main__":
models.Base.metadata.create_all(bind=engine)
init_test_db(session)
main()

38
menu.py
View File

@@ -31,11 +31,17 @@ class CommercialMenu:
filter :
contract: signed
"""
def __init__(self, customer_tools, contract_tools, event_tools, tools):
def __init__(self,
customer_tools,
contract_tools,
event_tools,
tools,
user_id):
self.customer_tools = customer_tools
self.contract_tools = contract_tools
self.event_tools = event_tools
self.tools = tools
self.user_id = user_id
self.prompt = Prompt()
def launch(self):
@@ -59,11 +65,15 @@ class CommercialMenu:
:return: exec the update tool with the chosen id
"""
options = {}
for item in self.tools.list(Customer):
for item in self.tools.filter(Customer,
("commercial_id", self.user_id)):
options[item[0].name] = item[0].id
choice = self.prompt.return_menu(options)
self.customer_tools.update(choice)
def customer_to_create(self):
self.customer_tools.create(self.user_id)
def customer_menu(self):
"""
display the CRUD menu for customer and get choice
@@ -71,7 +81,7 @@ class CommercialMenu:
"""
customer_options = {
"List": self.customer_tools.list,
"Create": self.customer_tools.create,
"Create": self.customer_to_create,
"Update": self.customers_for_update,
"Delete": self.customer_tools.delete,
}
@@ -84,7 +94,7 @@ class CommercialMenu:
:return: exec the update tool with the chosen id
"""
options = {}
for item in self.tools.list(Contract):
for item in self.tools.filter(Contract, ("commercial_id", self.user_id)):
options["Contrat "+str(item[0].id)] = item[0].id
choice = self.prompt.return_menu(options)
self.contract_tools.update(choice)
@@ -138,12 +148,15 @@ class ManagementMenu:
def __init__(self, collaborator_tools,
customer_tools,
contract_tools,
event_tools, tools):
event_tools,
tools, user_id):
self.collaborator_tools = collaborator_tools
self.customer_tools = customer_tools
self.contract_tools = contract_tools
self.event_tools = event_tools
self.tools = tools
self.user_id = user_id
self.prompt = Prompt()
def launch(self):
@@ -225,13 +238,24 @@ class ManagementMenu:
self.contract_tools.update(choice, customer_options,
commercial_options, event_options)
def contract_to_create(self):
(customer_options,
commercial_options,
event_options) = {}, {}, {}
commercial = self.collaborator_tools.get_by_team_id(1)
for customer in self.tools.list(Customer):
customer_options[customer[0].name] = customer[0].id
for user in commercial:
commercial_options[user[0].name] = user[0].id
self.contract_tools.create(customer_options, commercial_options)
def contract_menu(self):
"""
display the CRUD menu for contract and get choice
:return: exec the function associated with chosen item
"""
contract_options = {"List": self.contract_tools.list,
"Create": self.contract_tools.create,
"Create": self.contract_to_create,
"Update": self.contracts_for_update,
}
self.prompt.exec_menu(contract_options)
@@ -262,7 +286,7 @@ class ManagementMenu:
self.prompt.exec_menu(event_list_options)
def event_no_support(self):
self.event_tools.filter("support_id", "NULL")
self.event_tools.filter("support_id", None)
def event_menu(self):
event_options = {"List": self.event_list,

View File

@@ -59,8 +59,10 @@ class Customer(Base):
email: Mapped[str] = mapped_column(String(20))
phone: Mapped[int] = mapped_column(Integer)
company: Mapped[str] = mapped_column(String(40))
creation_date: Mapped[date] = mapped_column(Date,
server_default=func.now())
creation_date: Mapped[date] = mapped_column(DateTime,
default=datetime.now())
# creation_date: Mapped[date] = mapped_column(Date,
# server_default=func.now())
last_update: Mapped[Optional[datetime]] = mapped_column(DateTime)
commercial_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("collaborator.id"))
@@ -82,6 +84,8 @@ class Contract(Base):
id: Mapped[int] = mapped_column(primary_key=True)
signed: Mapped[bool] = mapped_column(Boolean, default=False)
# creation_date: Mapped[date] = mapped_column(Date,
# server_default=func.now())
creation_date: Mapped[date] = mapped_column(Date,
server_default=func.now())
amount: Mapped[int] = mapped_column(Integer)

View File

View File

@@ -0,0 +1,11 @@
class TestPasswordTool:
def test_check_wrong_user(self):
pass
def test_check_wrong_pwd(self):
pass
def get_by_name(self):
pass

77
tests/conftest.py Normal file
View File

@@ -0,0 +1,77 @@
import pytest
from models import Base, Credentials, Collaborator, Customer, Contract, Event
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from passlib.hash import argon2
DB_URL = "sqlite:///:memory:"
engine = create_engine(DB_URL, echo=False)
SessionLocal = sessionmaker(bind=engine)
cust1 = ("Cust1", "aa", 11, "Cust1CO")
cust2 = ("Cust2", "bb", 22, "Cust2CO")
@pytest.fixture
def session():
if not engine.url.get_backend_name() == "sqlite":
raise RuntimeError("Use SQLite backend to run tests\n"
"with command :\n"
"DB_URL=sqlite:///:memory: pytest -s -v .")
Base.metadata.create_all(engine)
try:
with SessionLocal() as session:
yield session
finally:
Base.metadata.drop_all(engine)
@pytest.fixture
def seed(session):
session.add_all(
[
Customer(name="Cust1", email="aa", phone=11, company="Cust1CO"),
Customer(name="Cust2", email="bb", phone=22, company="Cust2CO"),
]
)
session.commit()
session.add_all(
[
Collaborator(name="Col1", email="aa", phone=1, team_id=1),
Collaborator(name="Col2", email="bb", phone=2, team_id=2),
Collaborator(name="Col3", email="cc", phone=3, team_id=3),
Collaborator(name="Col4", email="dd", phone=4, team_id=2),
]
)
# session.add_all(
# [
# Collaborator(name="Com", email="a", phone=1, team_id=1),
# Collaborator(name="Man", email="b", phone=2, team_id=2),
# Collaborator(name="Sup", email="c", phone=3, team_id=3),
# Customer(name="Cust1", email="aa", phone=11, company="Cust1CO"),
# Customer(name="Cust2", email="bb", phone=22, company="Cust2CO"),
# ]
# )
# session.commit()
# session.add_all(
# [
# Credentials(collaborator_id=1,
# password_hash=argon2.hash("test")),
# Credentials(collaborator_id=2,
# password_hash=argon2.hash("test")),
# Credentials(collaborator_id=3,
# password_hash=argon2.hash("test")),
# Contract(signed=0, amount=200000, customer_id=1, commercial_id=1),
# ]
# )
# session.commit()
# session.add_all(
# [
# Event(name="Event1", customer_contact="Test",
# date_start="01.01.01", date_end="02.01.01",
# location=".",contract_id=1, customer_id=1),
# ]
# )
# session.commit()

0
tests/tools/__init__.py Normal file
View File

59
tests/tools/test_tools.py Normal file
View File

@@ -0,0 +1,59 @@
import pytest
from sqlalchemy.util import monkeypatch_proxied_specials
from models import Customer
from tools import CustomerTools, CollaboratorTools
class TestCustomerTools:
def test_db_should_be_populated(self, seed, session):
test = session.query(Customer).all()
assert test != None
def test_list_should_return_all_customers(self, seed, session):
tools = CustomerTools(session)
users = tools.list()
assert len(users) == 2
# def test_should_create_customer(self, seed, session, monkeypatch):
# CustomerTools(session).create(1)
# pass
def test_delete_user_should_remove_from_db(self, seed, session):
pass
class TestCollaboratorTools:
def test_should_reply_id_by_name(self, seed, session):
tool = CollaboratorTools(session)
reply = tool.get_id_by_name("Col1")
assert reply == 1
def test_should_reply_id_by_team_id(self, seed, session):
tool = CollaboratorTools(session)
reply = tool.get_by_team_id(1)
assert len(reply) == 1
reply = tool.get_by_team_id(2)
assert len(reply) == 2
def test_should_reply_team_id_by_name(self, seed, session):
tool = CollaboratorTools(session)
reply = tool.get_team_by_name("Col1")
reply2 = tool.get_team_by_name("Col2")
assert reply == 1
assert reply2 == 2
class TestPasswordTools:
def test_should_retrieve_hashed_password_by_username(self, seed, session):
pass
def test_right_user_could_connect(self, seed, session):
pass
def test_wrong_password_should_fail(self, seed, session):
pass
def test_unknown_user_should_fail(self, seed, session):
pass

View File

@@ -4,23 +4,43 @@ from authentication import PasswordTools
from views import View
from sqlalchemy.orm import Session
from sqlalchemy import select, update, insert, delete
from datetime import datetime, date
class Tools:
"""
Generic tools with object as an arg
"""
def __init__(self, db: Session):
self.db = db
self.view = View()
def list(self, object):
"""
Select all objects from DB and return the list
:param object: object in Collaborator, Customer, Contract, Event
:return: list of object
"""
while self.db:
return self.db.execute(select(object)).all()
def filter(self, object, filter):
"""
Select objects from DB where filter and return the list
:param object: object in Collaborator, Customer, Contract, Event
:param filter: (attribute, value):tuple
:return: list of selected object matching filter
"""
item, value = filter
stmt = (select(object).where(**{item: value}))
print(stmt)
# while self.db:
# return self.db.execute(select(object).where(**{item: value})).all()
while self.db:
result = self.db.execute(
select(object).where(**{item: value})).all()
if not result:
self.view.display_error()
self.view.display_results(result)
class CollaboratorTools:
"""
@@ -34,6 +54,8 @@ class CollaboratorTools:
def get_id_by_name(self, username):
collaborator = self.db.execute(
select(Collaborator).where(Collaborator.name == username)).scalar()
if not collaborator:
return None
return collaborator.id
def get_by_team_id(self, team_id):
@@ -111,7 +133,7 @@ class CollaboratorTools:
ret = self.db.execute(
select(Collaborator).where(Collaborator.name == username)).scalar()
if ret is None:
print({'message': "This username doesn't exist"})
self.view.display_error()
return ret
else:
return ret.team_id
@@ -139,7 +161,7 @@ class CustomerTools:
self.view.display_results(result)
return result
def create(self) -> None:
def create(self, user_id) -> None:
"""
Create a new customer with minimum information
:return: None; creates object in DB
@@ -150,6 +172,7 @@ class CustomerTools:
email=customer['email'],
phone=customer['phone'],
company=customer['company'],
commercial_id=user_id,
)
self.db.add(new_customer)
self.db.commit()
@@ -164,7 +187,7 @@ class CustomerTools:
cust = self.db.get(Customer, my_id)
item, value = self.view.prompt_for_customer_update()
stmt = (update(Customer).where(Customer.id == my_id).values(
**{item: value}))
**{item: value}, last_update=datetime.now()))
self.db.execute(stmt)
self.db.commit()
self.view.display_change(cust.name, item, value)
@@ -190,10 +213,12 @@ class CustomerTools:
ret = self.db.execute(
select(Customer).where(Customer.commercial_id == my_id)).all()
if ret is None:
print({'message': "No customer found"})
# print({'message': "No customer found"})
self.view.display_error()
return ret
else:
print({'message': "No commercial with this id"})
# print({'message': "No commercial with this id"})
self.view.display_error()
return None
@@ -231,15 +256,20 @@ class ContractTools:
select(Contract).where(Contract.signed == 0))
self.view.display_results(result)
def create(self) -> None:
def create(self,
customer_options,
commercial_options) -> None:
"""
Create a new contracts with minimum information
:return: None; creates object in DB
"""
contract = self.view.prompt_for_contract()
contract = self.view.prompt_for_contract(customer_options,
commercial_options)
if not contract['amount']:
contract['amount'] = 0
new_contract = Contract(
customer=contract['customer'],
commercial=contract['commercial'],
customer_id=contract['customer'],
commercial_id=contract['commercial'],
amount=contract['amount'],
)
self.db.add(new_contract)
@@ -310,7 +340,9 @@ class EventTools:
"""
result = self.db.execute(
select(Event).filter_by(**{field: value})).all()
print(field, value, result)
if not result:
self.view.display_error()
else:
self.view.display_results(result)
def filter_owned(self, user_id):

View File

@@ -78,11 +78,13 @@ class View:
}
return self.prompt_for_update(options)
def prompt_for_contract(self) -> dict:
def prompt_for_contract(self, customer_options, commercial_options) -> dict:
contract = {}
print("** New contract **")
contract['customer'] = input("Customer (id) ? : ")
contract['commercial'] = input("Commercial (id) ")
print("Customer ? :")
contract['customer'] = return_menu(customer_options)
print("Commercial ? :")
contract['commercial'] = return_menu(commercial_options)
contract['amount'] = input("Budget ? : ")
return contract
@@ -135,5 +137,11 @@ class View:
def display_error(self):
print("No object matches this query")
def display_no_user(self):
print("This user doesn't exist")
def display_co_failed(self):
print("Connexion failed.")
def display_items(self):
print()