mirror of
https://gitflic.ru/project/maks1ms/ocab.git
synced 2024-12-24 00:33:05 +03:00
Исправлены импорты внутри модулей, добавлены: модуль для исключений, скрипт инициализации путей, переделано подключение к бд
This commit is contained in:
parent
88db66c360
commit
1e7c70078b
1
.gitignore
vendored
1
.gitignore
vendored
@ -5,3 +5,4 @@ env
|
|||||||
venv
|
venv
|
||||||
__pycache__
|
__pycache__
|
||||||
OCAB.db
|
OCAB.db
|
||||||
|
paths.json
|
15
init.py
Normal file
15
init.py
Normal file
@ -0,0 +1,15 @@
|
|||||||
|
from pathlib import Path
|
||||||
|
from json import dumps
|
||||||
|
|
||||||
|
pwd = Path().cwd()
|
||||||
|
dir_core = pwd / "src" / "core"
|
||||||
|
dir_modules_standard = pwd / "src" / "modules" / "standard"
|
||||||
|
dir_modules_custom = pwd / "src" / "modules" / "custom"
|
||||||
|
|
||||||
|
json = {
|
||||||
|
'core': str(dir_core),
|
||||||
|
'modules standard': str(dir_modules_standard),
|
||||||
|
'modules custom': str(dir_modules_custom),
|
||||||
|
}
|
||||||
|
with open("paths.json", "w", encoding="utf8") as f:
|
||||||
|
f.write(dumps(json, indent=4))
|
4
run_tests
Normal file
4
run_tests
Normal file
@ -0,0 +1,4 @@
|
|||||||
|
#! /bin/sh
|
||||||
|
cd src
|
||||||
|
python -m unittest discover -v
|
||||||
|
cd ..
|
@ -0,0 +1 @@
|
|||||||
|
import service
|
@ -1,18 +1,22 @@
|
|||||||
from src.modules.standart.config.config import yaml_load
|
from src.modules.standard.config.config import get_config
|
||||||
|
from src.modules.standard.database.models.base import database
|
||||||
|
from src.modules.standard.database.api import connect_database
|
||||||
|
|
||||||
from asyncio import run
|
from asyncio import run
|
||||||
from aiogram import Bot, Dispatcher
|
from aiogram import Bot, Dispatcher
|
||||||
|
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
try:
|
try:
|
||||||
bot = Bot(token=yaml_load["TELEGRAM"]["TOKEN"])
|
database = connect_database()
|
||||||
|
bot = Bot(token=get_config()["TELEGRAM"]["TOKEN"])
|
||||||
|
|
||||||
dp = Dispatcher()
|
dp = Dispatcher()
|
||||||
|
|
||||||
await dp.start_polling(bot)
|
await dp.start_polling(bot)
|
||||||
|
|
||||||
finally:
|
finally:
|
||||||
await bot.session.close()
|
await bot.session.close()
|
||||||
|
database.close()
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
1
src/modules/standard/__init__.py
Normal file
1
src/modules/standard/__init__.py
Normal file
@ -0,0 +1 @@
|
|||||||
|
from . import config, database, exceptions, roles
|
1
src/modules/standard/config/__init__.py
Normal file
1
src/modules/standard/config/__init__.py
Normal file
@ -0,0 +1 @@
|
|||||||
|
from . import config
|
15
src/modules/standard/config/config.py
Normal file
15
src/modules/standard/config/config.py
Normal file
@ -0,0 +1,15 @@
|
|||||||
|
import yaml
|
||||||
|
from service import paths
|
||||||
|
|
||||||
|
|
||||||
|
def get_config(is_test: bool = False) -> dict:
|
||||||
|
if is_test:
|
||||||
|
path = f"{paths.modules_standard}/config/tests"
|
||||||
|
else:
|
||||||
|
path = paths.core
|
||||||
|
path = f"{path}/config.yaml"
|
||||||
|
|
||||||
|
with open(path, 'r') as file:
|
||||||
|
return yaml.full_load(file)
|
||||||
|
|
||||||
|
|
@ -1,6 +1,8 @@
|
|||||||
from src.modules.standart.config.config import yaml_load
|
from src.modules.standard.config.config import get_config
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
|
yaml_load = get_config(is_test=True)
|
||||||
|
|
||||||
|
|
||||||
class TestConfig(unittest.TestCase):
|
class TestConfig(unittest.TestCase):
|
||||||
def test_yaml_load_correctness(self):
|
def test_yaml_load_correctness(self):
|
1
src/modules/standard/database/__init__.py
Normal file
1
src/modules/standard/database/__init__.py
Normal file
@ -0,0 +1 @@
|
|||||||
|
from . import api, models
|
@ -1,9 +1,26 @@
|
|||||||
from src.modules.standart.database.models.chats import Chats
|
from .models.chats import Chats
|
||||||
from src.modules.standart.database.models.messages import Messages
|
from .models.messages import Messages
|
||||||
from src.modules.standart.database.models.users import Users
|
from .models.users import Users
|
||||||
|
from service import paths
|
||||||
|
from ..exceptions.module_exceptions import MissingModuleName, NotExpectedModuleName
|
||||||
import peewee as pw
|
import peewee as pw
|
||||||
|
|
||||||
|
def connect_database(is_test: bool = False, module: str | None = None):
|
||||||
|
if is_test:
|
||||||
|
if not module:
|
||||||
|
raise MissingModuleName()
|
||||||
|
db_path = f"{paths.modules_standard}/{module}/tests/database"
|
||||||
|
else:
|
||||||
|
if module:
|
||||||
|
raise NotExpectedModuleName()
|
||||||
|
db_path = f"{paths.core}/database"
|
||||||
|
|
||||||
|
_database = pw.SqliteDatabase(f"{db_path}/OCAB.db")
|
||||||
|
Chats._meta.database = _database
|
||||||
|
Messages._meta.database = _database
|
||||||
|
Users._meta.database = _database
|
||||||
|
|
||||||
|
return _database, f"{db_path}/OCAB.db"
|
||||||
|
|
||||||
def create_tables(db: pw.SqliteDatabase):
|
def create_tables(db: pw.SqliteDatabase):
|
||||||
"""Создание таблиц"""
|
"""Создание таблиц"""
|
1
src/modules/standard/database/models/__init__.py
Normal file
1
src/modules/standard/database/models/__init__.py
Normal file
@ -0,0 +1 @@
|
|||||||
|
|
@ -1,9 +1,9 @@
|
|||||||
from .base import BaseModel
|
|
||||||
|
|
||||||
import peewee as pw
|
import peewee as pw
|
||||||
|
|
||||||
|
|
||||||
class Chats(BaseModel):
|
class Chats(pw.Model):
|
||||||
|
class Meta:
|
||||||
|
...
|
||||||
chat_role = pw.IntegerField(null=False, default=1)
|
chat_role = pw.IntegerField(null=False, default=1)
|
||||||
chat_stats = pw.IntegerField(null=False)
|
chat_stats = pw.IntegerField(null=False)
|
||||||
chat_federation = pw.IntegerField(null=False)
|
chat_federation = pw.IntegerField(null=False)
|
@ -1,9 +1,9 @@
|
|||||||
from .base import BaseModel
|
|
||||||
|
|
||||||
import peewee as pw
|
import peewee as pw
|
||||||
|
|
||||||
|
|
||||||
class Messages(BaseModel):
|
class Messages(pw.Model):
|
||||||
|
class Meta:
|
||||||
|
...
|
||||||
message_text = pw.TextField(null=False)
|
message_text = pw.TextField(null=False)
|
||||||
message_id_sender = pw.IntegerField(null=False)
|
message_id_sender = pw.IntegerField(null=False)
|
||||||
answer_to_id = pw.IntegerField(null=True)
|
answer_to_id = pw.IntegerField(null=True)
|
@ -1,9 +1,9 @@
|
|||||||
from .base import BaseModel
|
|
||||||
|
|
||||||
import peewee as pw
|
import peewee as pw
|
||||||
|
|
||||||
|
|
||||||
class Users(BaseModel):
|
class Users(pw.Model):
|
||||||
|
class Meta:
|
||||||
|
...
|
||||||
user_name = pw.CharField(null=False) # до 255 символов
|
user_name = pw.CharField(null=False) # до 255 символов
|
||||||
user_tag = pw.CharField(null=True)
|
user_tag = pw.CharField(null=True)
|
||||||
user_role = pw.IntegerField(null=True)
|
user_role = pw.IntegerField(null=True)
|
@ -1,12 +1,24 @@
|
|||||||
import unittest
|
import unittest
|
||||||
|
import os
|
||||||
|
|
||||||
from src.modules.standart.database.api import *
|
from ..api import *
|
||||||
from src.modules.standart.database.models.base import database
|
from ...exceptions.module_exceptions import MissingModuleName, NotExpectedModuleName
|
||||||
|
|
||||||
create_tables(database)
|
|
||||||
|
|
||||||
|
|
||||||
class TestDatabaseAPI(unittest.TestCase):
|
class TestDatabaseAPI(unittest.TestCase):
|
||||||
|
database = None
|
||||||
|
path = None
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def setUpClass(cls):
|
||||||
|
cls.database, cls.path = connect_database(is_test=True, module="database")
|
||||||
|
create_tables(cls.database)
|
||||||
|
def test_fail_connect(cls):
|
||||||
|
with cls.assertRaises(MissingModuleName):
|
||||||
|
cls.database, cls.path = connect_database(is_test=True)
|
||||||
|
|
||||||
|
with cls.assertRaises(NotExpectedModuleName):
|
||||||
|
cls.database, cls.path = connect_database(module="database")
|
||||||
|
|
||||||
def test_add_and_get_chat(self):
|
def test_add_and_get_chat(self):
|
||||||
add_chat(chat_id=21, chat_role=0, chat_stats=0, chat_federation=0)
|
add_chat(chat_id=21, chat_role=0, chat_stats=0, chat_federation=0)
|
||||||
@ -71,6 +83,11 @@ class TestDatabaseAPI(unittest.TestCase):
|
|||||||
updated_user2 = get_user(105)
|
updated_user2 = get_user(105)
|
||||||
self.assertEqual(updated_user2.user_name, "NewName2")
|
self.assertEqual(updated_user2.user_name, "NewName2")
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def tearDownClass(cls):
|
||||||
|
cls.database.close()
|
||||||
|
os.system(f"rm {cls.path}")
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
unittest.main()
|
unittest.main()
|
1
src/modules/standard/exceptions/__init__.py
Normal file
1
src/modules/standard/exceptions/__init__.py
Normal file
@ -0,0 +1 @@
|
|||||||
|
from . import module_exceptions
|
6
src/modules/standard/exceptions/info.json
Normal file
6
src/modules/standard/exceptions/info.json
Normal file
@ -0,0 +1,6 @@
|
|||||||
|
{
|
||||||
|
"name": "Exceptions",
|
||||||
|
"description": "Модуль с исключениями",
|
||||||
|
"author": "OCAB Team",
|
||||||
|
"version": "1.0"
|
||||||
|
}
|
13
src/modules/standard/exceptions/module_exceptions.py
Normal file
13
src/modules/standard/exceptions/module_exceptions.py
Normal file
@ -0,0 +1,13 @@
|
|||||||
|
class MissingModuleName(BaseException):
|
||||||
|
def __init__(self):
|
||||||
|
self.message = "Пропущено название директории модуля"
|
||||||
|
|
||||||
|
super().__init__(self.message)
|
||||||
|
|
||||||
|
|
||||||
|
class NotExpectedModuleName(BaseException):
|
||||||
|
def __init__(self):
|
||||||
|
self.message = "Не ожидалось название директории модуля"
|
||||||
|
|
||||||
|
super().__init__(self.message)
|
||||||
|
|
@ -1,7 +1,7 @@
|
|||||||
from src.modules.standart.database import get_user_role
|
from ..database.api import get_user_role
|
||||||
from src.modules.standart.config.config import yaml_load
|
from ..config.config import get_config
|
||||||
|
|
||||||
|
|
||||||
|
yaml_load = get_config()
|
||||||
class Roles:
|
class Roles:
|
||||||
user = "USER"
|
user = "USER"
|
||||||
moderator = "MODERATOR"
|
moderator = "MODERATOR"
|
50
src/modules/standard/roles/tests/test_roles.py
Normal file
50
src/modules/standard/roles/tests/test_roles.py
Normal file
@ -0,0 +1,50 @@
|
|||||||
|
import os
|
||||||
|
import unittest
|
||||||
|
from ...database.api import create_tables, connect_database, add_user
|
||||||
|
from ..api import Roles
|
||||||
|
|
||||||
|
|
||||||
|
class TestRoles(unittest.IsolatedAsyncioTestCase):
|
||||||
|
database = None
|
||||||
|
path = None
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def setUpClass(cls):
|
||||||
|
cls.roles = Roles()
|
||||||
|
cls.database, cls.path = connect_database(is_test=True, module="roles")
|
||||||
|
create_tables(cls.database)
|
||||||
|
|
||||||
|
add_user(user_id=1, user_name="TestUser1", user_tag="TestTag1", user_role=0, user_stats=30, user_rep=15)
|
||||||
|
add_user(user_id=2, user_name="TestUser3", user_tag="TestTag3", user_role=1, user_stats=30, user_rep=15)
|
||||||
|
add_user(user_id=3, user_name="TestUser4", user_tag="TestTag4", user_role=2, user_stats=30, user_rep=15)
|
||||||
|
add_user(user_id=4, user_name="TestUser2", user_tag="TestTag2", user_role=3, user_stats=30, user_rep=15)
|
||||||
|
async def test_check_admin_permission(cls):
|
||||||
|
cls.assertTrue(await cls.roles.check_admin_permission(1))
|
||||||
|
cls.assertFalse(await cls.roles.check_admin_permission(2))
|
||||||
|
cls.assertFalse(await cls.roles.check_admin_permission(3))
|
||||||
|
cls.assertFalse(await cls.roles.check_admin_permission(4))
|
||||||
|
|
||||||
|
async def test_check_moderator_permission(cls):
|
||||||
|
cls.assertTrue(await cls.roles.check_moderator_permission(2))
|
||||||
|
cls.assertFalse(await cls.roles.check_moderator_permission(0))
|
||||||
|
cls.assertFalse(await cls.roles.check_moderator_permission(1))
|
||||||
|
cls.assertFalse(await cls.roles.check_moderator_permission(3))
|
||||||
|
|
||||||
|
async def test_get_role_name(cls):
|
||||||
|
cls.assertEqual(await cls.roles.get_role_name(cls.roles.admin_role_id), "ADMIN")
|
||||||
|
cls.assertEqual(await cls.roles.get_role_name(cls.roles.moderator_role_id), "MODERATOR")
|
||||||
|
cls.assertEqual(await cls.roles.get_role_name(cls.roles.user_role_id), "USER")
|
||||||
|
cls.assertEqual(await cls.roles.get_role_name(cls.roles.bot_role_id), "BOT")
|
||||||
|
with cls.assertRaises(ValueError):
|
||||||
|
await cls.roles.get_role_name(999) # Несуществующий ID роли
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def tearDownClass(cls):
|
||||||
|
cls.database.close()
|
||||||
|
os.system(f"rm {cls.path}")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
@ -1,5 +0,0 @@
|
|||||||
import yaml
|
|
||||||
|
|
||||||
|
|
||||||
with open('config.yaml', 'r') as file:
|
|
||||||
yaml_load = yaml.full_load(file)
|
|
@ -1,3 +0,0 @@
|
|||||||
from .api import *
|
|
||||||
from .models import *
|
|
||||||
from .models.base import database
|
|
@ -1 +0,0 @@
|
|||||||
from .base import BaseModel
|
|
@ -1,8 +0,0 @@
|
|||||||
import peewee as pw
|
|
||||||
|
|
||||||
database = pw.SqliteDatabase("database/OCAB.db")
|
|
||||||
|
|
||||||
|
|
||||||
class BaseModel(pw.Model):
|
|
||||||
class Meta:
|
|
||||||
database = database
|
|
@ -1,35 +0,0 @@
|
|||||||
import unittest
|
|
||||||
from src.modules.standart.database.api import create_tables, add_user
|
|
||||||
from src.modules.standart.database.models.base import database
|
|
||||||
from src.modules.standart.roles.api import Roles
|
|
||||||
import asyncio
|
|
||||||
|
|
||||||
create_tables(database)
|
|
||||||
|
|
||||||
class TestRoles(unittest.IsolatedAsyncioTestCase):
|
|
||||||
async def asyncSetUp(self):
|
|
||||||
self.roles = Roles()
|
|
||||||
|
|
||||||
async def test_check_admin_permission(self):
|
|
||||||
self.assertTrue(await self.roles.check_admin_permission(1))
|
|
||||||
self.assertFalse(await self.roles.check_admin_permission(2))
|
|
||||||
self.assertFalse(await self.roles.check_admin_permission(3))
|
|
||||||
self.assertFalse(await self.roles.check_admin_permission(4))
|
|
||||||
|
|
||||||
async def test_check_moderator_permission(self):
|
|
||||||
self.assertTrue(await self.roles.check_moderator_permission(2))
|
|
||||||
self.assertFalse(await self.roles.check_moderator_permission(0))
|
|
||||||
self.assertFalse(await self.roles.check_moderator_permission(1))
|
|
||||||
self.assertFalse(await self.roles.check_moderator_permission(3))
|
|
||||||
|
|
||||||
async def test_get_role_name(self):
|
|
||||||
self.assertEqual(await self.roles.get_role_name(self.roles.admin_role_id), "ADMIN")
|
|
||||||
self.assertEqual(await self.roles.get_role_name(self.roles.moderator_role_id), "MODERATOR")
|
|
||||||
self.assertEqual(await self.roles.get_role_name(self.roles.user_role_id), "USER")
|
|
||||||
self.assertEqual(await self.roles.get_role_name(self.roles.bot_role_id), "BOT")
|
|
||||||
with self.assertRaises(ValueError):
|
|
||||||
await self.roles.get_role_name(999) # Несуществующий ID роли
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
|
||||||
unittest.main()
|
|
23
src/service.py
Normal file
23
src/service.py
Normal file
@ -0,0 +1,23 @@
|
|||||||
|
import os.path
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from json import loads
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Path:
|
||||||
|
core: str
|
||||||
|
modules_standard: str
|
||||||
|
modules_custom: str
|
||||||
|
|
||||||
|
def _get_paths(path_to_json: str):
|
||||||
|
with open(path_to_json, encoding="utf8") as f:
|
||||||
|
paths = loads(f.read())
|
||||||
|
return Path(
|
||||||
|
core=paths["core"],
|
||||||
|
modules_standard=paths["modules standard"],
|
||||||
|
modules_custom=paths["modules custom"]
|
||||||
|
)
|
||||||
|
|
||||||
|
cwd = os.getcwd()
|
||||||
|
cwd = cwd[:cwd.index('/src')]
|
||||||
|
|
||||||
|
paths = _get_paths(f"{cwd}/paths.json")
|
Loading…
Reference in New Issue
Block a user