diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 3aad913..918cd75 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -40,6 +40,12 @@ jobs: - name: Run linter run: git diff --name-only HEAD~10 HEAD | xargs pre-commit run --files + - name: Run tests + run: | + modified_files=$(git diff --name-only HEAD~10 HEAD | tr '\n' ',') + modified_files=${modified_files%,} + python -m unittest discover -p "test_*.py" -v --start-directory src/tests/ --pattern="${modified_files}" + - name: Release env: GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 00799ce..80af900 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -21,7 +21,7 @@ repos: - repo: https://github.com/commitizen-tools/commitizen - rev: 3.2.2 + rev: v3.2.2 hooks: - id: commitizen stages: [ commit-msg ] diff --git a/CHANGELOG.md b/CHANGELOG.md index 724ece1..6076287 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,12 +1,5 @@ # Semantic Versioning Changelog -# [1.18.0](https://github.com/casdoor/casdoor-python-sdk/compare/v1.17.2...v1.18.0) (2023-11-30) - - -### Features - -* add test cases for organization and group ([#76](https://github.com/casdoor/casdoor-python-sdk/issues/76)) ([4319243](https://github.com/casdoor/casdoor-python-sdk/commit/4319243429ae48feb80de0bbb05f3078e36afd54)) - ## [1.17.2](https://github.com/casdoor/casdoor-python-sdk/compare/v1.17.1...v1.17.2) (2023-11-24) diff --git a/src/casdoor/adapter.py b/src/casdoor/adapter.py index 8316341..353893f 100644 --- a/src/casdoor/adapter.py +++ b/src/casdoor/adapter.py @@ -34,6 +34,27 @@ def __init__(self): self.tableNamePrefix = "" self.isEnabled = False + @classmethod + def new(cls, owner, name, created_time, host, user): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.host = host + self.user = user + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + adapter = cls() + for key, value in data.items(): + if hasattr(adapter, key): + setattr(adapter, key, value) + return adapter + def __str__(self): return str(self.__dict__) @@ -42,7 +63,7 @@ def to_dict(self) -> dict: class _AdapterSDK: - def get_adapters(self) -> List[Dict]: + def get_adapters(self) -> List[Adapter]: """ Get the adapters from Casdoor. @@ -55,10 +76,15 @@ def get_adapters(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - adapters = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + adapters = [] + for adapter in response["data"]: + adapters.append(Adapter.from_dict(adapter)) return adapters - def get_adapter(self, adapter_id: str) -> Dict: + def get_adapter(self, adapter_id: str) -> Adapter: """ Get the adapter from Casdoor providing the adapter_id. @@ -72,13 +98,14 @@ def get_adapter(self, adapter_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - adapter = r.json() - return adapter + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Adapter.from_dict(response["data"]) - def modify_adapter(self, method: str, adapter: Adapter) -> Dict: + def modify_adapter(self, method: str, adapter: Adapter) -> str: url = self.endpoint + f"/api/{method}" - if adapter.owner == "": - adapter.owner = self.org_name + adapter.owner = self.org_name params = { "id": f"{adapter.owner}/{adapter.name}", "clientId": self.client_id, @@ -87,7 +114,9 @@ def modify_adapter(self, method: str, adapter: Adapter) -> Dict: adapter_info = json.dumps(adapter.to_dict()) r = requests.post(url, params=params, data=adapter_info) response = r.json() - return response + if response["status"] != "ok": + raise Exception(response["msg"]) + return str(response["data"]) def add_adapter(self, adapter: Adapter) -> Dict: response = self.modify_adapter("add-adapter", adapter) diff --git a/src/casdoor/cert.py b/src/casdoor/cert.py index 942fc2a..800e36c 100644 --- a/src/casdoor/cert.py +++ b/src/casdoor/cert.py @@ -13,7 +13,7 @@ # limitations under the License. import json -from typing import Dict, List +from typing import List import requests @@ -34,6 +34,31 @@ def __init__(self): self.authorityPublicKey = "" self.authorityRootPublicKey = "" + @classmethod + def new(cls, owner, name, created_time, display_name, scope, type, crypto_algorithm, bit_size, expire_in_years): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.scope = scope + self.type = type + self.cryptoAlgorithm = crypto_algorithm + self.bitSize = bit_size + self.expireInYears = expire_in_years + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + cert = cls() + for key, value in data.items(): + if hasattr(cert, key): + setattr(cert, key, value) + return cert + def __str__(self): return str(self.__dict__) @@ -42,7 +67,7 @@ def to_dict(self) -> dict: class _CertSDK: - def get_certs(self) -> List[Dict]: + def get_certs(self) -> List[Cert]: """ Get the certs from Casdoor. @@ -55,10 +80,16 @@ def get_certs(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - certs = r.json() - return certs + response = r.json() + if response["status"] != "ok": + raise ValueError(response["msg"]) - def get_cert(self, cert_id: str) -> Dict: + res = [] + for element in response["data"]: + res.append(Cert.from_dict(element)) + return res + + def get_cert(self, cert_id: str) -> Cert: """ Get the cert from Casdoor providing the cert_id. @@ -72,13 +103,15 @@ def get_cert(self, cert_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - cert = r.json() - return cert + response = r.json() + if response["status"] != "ok": + raise ValueError(response["msg"]) - def modify_cert(self, method: str, cert: Cert) -> Dict: + return Cert.from_dict(response["data"]) + + def modify_cert(self, method: str, cert: Cert) -> str: url = self.endpoint + f"/api/{method}" - if cert.owner == "": - cert.owner = self.org_name + cert.owner = self.org_name params = { "id": f"{cert.owner}/{cert.name}", "clientId": self.client_id, @@ -87,16 +120,18 @@ def modify_cert(self, method: str, cert: Cert) -> Dict: cert_info = json.dumps(cert.to_dict()) r = requests.post(url, params=params, data=cert_info) response = r.json() - return response + if response["status"] != "ok": + raise ValueError(response["msg"]) + return str(response["data"]) - def add_cert(self, cert: Cert) -> Dict: + def add_cert(self, cert: Cert) -> str: response = self.modify_cert("add-cert", cert) return response - def update_cert(self, cert: Cert) -> Dict: + def update_cert(self, cert: Cert) -> str: response = self.modify_cert("update-cert", cert) return response - def delete_cert(self, cert: Cert) -> Dict: + def delete_cert(self, cert: Cert) -> str: response = self.modify_cert("delete-cert", cert) return response diff --git a/src/casdoor/enforcer.py b/src/casdoor/enforcer.py index c608934..e076f7b 100644 --- a/src/casdoor/enforcer.py +++ b/src/casdoor/enforcer.py @@ -30,6 +30,29 @@ def __init__(self): self.adapter = "" self.isEnabled = False + @classmethod + def new(cls, owner, name, created_time, display_name, description, model, adapter): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.description = description + self.model = model + self.adapter = adapter + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + enforcer = cls() + for key, value in data.items(): + if hasattr(enforcer, key): + setattr(enforcer, key, value) + return enforcer + def __str__(self): return str(self.__dict__) @@ -51,7 +74,12 @@ def get_enforcers(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - enforcers = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + enforcers = [] + for enforcer in response["data"]: + enforcers.append(Enforcer.from_dict(enforcer)) return enforcers def get_enforcer(self, enforcer_id: str) -> Dict: @@ -68,13 +96,15 @@ def get_enforcer(self, enforcer_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - enforcer = r.json() - return enforcer + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + + return Enforcer.from_dict(response["data"]) def modify_enforcer(self, method: str, enforcer: Enforcer) -> Dict: url = self.endpoint + f"/api/{method}" - if enforcer.owner == "": - enforcer.owner = self.org_name + enforcer.owner = self.org_name params = { "id": f"{enforcer.owner}/{enforcer.name}", "clientId": self.client_id, diff --git a/src/casdoor/main.py b/src/casdoor/main.py index bf9b763..9089389 100644 --- a/src/casdoor/main.py +++ b/src/casdoor/main.py @@ -27,7 +27,7 @@ from .model import _ModelSDK from .organization import _OrganizationSDK from .payment import _PaymentSDK -from .permisssion import _PermissionSDK +from .permission import _PermissionSDK from .plan import _PlanSDK from .pricing import _PricingSDK from .product import _ProductSDK diff --git a/src/casdoor/model.py b/src/casdoor/model.py index c79abb5..3704acc 100644 --- a/src/casdoor/model.py +++ b/src/casdoor/model.py @@ -36,8 +36,29 @@ def __init__(self): self.title = "" self.key = "" self.children = [Model] + self.modelText = "" self.isEnabled = False + @classmethod + def new(cls, owner, name, created_time, display_name, model_text): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.modelText = model_text + return self + + @classmethod + def from_dict(cls, data: dict): + if not data: + return None + model = cls() + for key, value in data.items(): + if hasattr(model, key): + setattr(model, key, value) + return model + def __str__(self): return str(self.__dict__) @@ -59,7 +80,12 @@ def get_models(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - models = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + models = [] + for model in response["data"]: + models.append(Model.from_dict(model)) return models def get_model(self, model_id: str) -> Dict: @@ -76,19 +102,20 @@ def get_model(self, model_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - model = r.json() - return model + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Model.from_dict(response["data"]) def modify_model(self, method: str, model: Model) -> Dict: url = self.endpoint + f"/api/{method}" - if model.owner == "": - model.owner = self.org_name + model.owner = self.org_name params = { "id": f"{model.owner}/{model.name}", "clientId": self.client_id, "clientSecret": self.client_secret, } - model_info = json.dumps(model.to_dict()) + model_info = json.dumps(model.to_dict(), default=self.custom_encoder) r = requests.post(url, params=params, data=model_info) response = r.json() return response @@ -104,3 +131,7 @@ def update_model(self, model: Model) -> Dict: def delete_model(self, model: Model) -> Dict: response = self.modify_model("delete-model", model) return response + + def custom_encoder(self, o): + if isinstance(o, (Model, User)): + return o.__dict__ diff --git a/src/casdoor/payment.py b/src/casdoor/payment.py index da40ae8..5c19a9d 100644 --- a/src/casdoor/payment.py +++ b/src/casdoor/payment.py @@ -48,6 +48,26 @@ def __init__(self): self.state = "" self.message = "" + @classmethod + def new(cls, owner, name, created_time, display_name, product_name): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.productName = product_name + return self + + @classmethod + def from_dict(cls, data: dict): + if not data: + return None + payment = cls() + for key, value in data.items(): + if hasattr(payment, key): + setattr(payment, key, value) + return payment + def __str__(self): return str(self.__dict__) @@ -69,7 +89,12 @@ def get_payments(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - payments = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + payments = [] + for payment in response["data"]: + payments.append(Payment.from_dict(payment)) return payments def get_payment(self, payment_id: str) -> Dict: @@ -86,13 +111,14 @@ def get_payment(self, payment_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - payment = r.json() - return payment + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Payment.from_dict(response["data"]) def modify_payment(self, method: str, payment: Payment) -> Dict: url = self.endpoint + f"/api/{method}" - if payment.owner == "": - payment.owner = self.org_name + payment.owner = self.org_name params = { "id": f"{payment.owner}/{payment.name}", "clientId": self.client_id, diff --git a/src/casdoor/permisssion.py b/src/casdoor/permission.py similarity index 67% rename from src/casdoor/permisssion.py rename to src/casdoor/permission.py index b31a393..b524ba3 100644 --- a/src/casdoor/permisssion.py +++ b/src/casdoor/permission.py @@ -40,6 +40,51 @@ def __init__(self): self.approveTime = "" self.state = "" + @classmethod + def new( + cls, + owner, + name, + created_time, + display_name, + description, + users, + roles, + domains, + model, + resource_type, + resources, + actions, + effect, + is_enabled, + ): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.description = description + self.users = users + self.roles = roles + self.domains = domains + self.model = model + self.resourceType = resource_type + self.resources = resources + self.actions = actions + self.effect = effect + self.isEnabled = is_enabled + return self + + @classmethod + def from_dict(cls, data: dict): + if not data: + return None + permission = cls() + for key, value in data.items(): + if hasattr(permission, key): + setattr(permission, key, value) + return permission + def __str__(self): return str(self.__dict__) @@ -61,7 +106,12 @@ def get_permissions(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - permissions = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + permissions = [] + for permission in response["data"]: + permissions.append(Permission.from_dict(permission)) return permissions def get_permission(self, permission_id: str) -> Dict: @@ -78,13 +128,14 @@ def get_permission(self, permission_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - permission = r.json() - return permission + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Permission.from_dict(response["data"]) def modify_permission(self, method: str, permission: Permission) -> Dict: url = self.endpoint + f"/api/{method}" - if permission.owner == "": - permission.owner = self.org_name + permission.owner = self.org_name params = { "id": f"{permission.owner}/{permission.name}", "clientId": self.client_id, @@ -93,7 +144,9 @@ def modify_permission(self, method: str, permission: Permission) -> Dict: permission_info = json.dumps(permission.to_dict()) r = requests.post(url, params=params, data=permission_info) response = r.json() - return response + if response["status"] != "ok": + raise Exception(response["msg"]) + return str(response["data"]) def add_permission(self, permission: Permission) -> Dict: response = self.modify_permission("add-permission", permission) diff --git a/src/casdoor/plan.py b/src/casdoor/plan.py index c56e4f6..290ae44 100644 --- a/src/casdoor/plan.py +++ b/src/casdoor/plan.py @@ -32,6 +32,27 @@ def __init__(self): self.role = "" self.options = [""] + @classmethod + def new(cls, owner, name, created_time, display_name, description): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.description = description + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + plan = cls() + for key, value in data.items(): + if hasattr(plan, key): + setattr(plan, key, value) + return plan + def __str__(self): return str(self.__dict__) @@ -53,7 +74,12 @@ def get_plans(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - plans = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + plans = [] + for plan in response["data"]: + plans.append(Plan.from_dict(plan)) return plans def get_plan(self, plan_id: str) -> Dict: @@ -70,13 +96,15 @@ def get_plan(self, plan_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - plan = r.json() - return plan + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + + return Plan.from_dict(response["data"]) def modify_plan(self, method: str, plan: Plan) -> Dict: url = self.endpoint + f"/api/{method}" - if plan.owner == "": - plan.owner = self.org_name + plan.owner = self.org_name params = { "id": f"{plan.owner}/{plan.name}", "clientId": self.client_id, @@ -85,6 +113,8 @@ def modify_plan(self, method: str, plan: Plan) -> Dict: plan_info = json.dumps(plan.to_dict()) r = requests.post(url, params=params, data=plan_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_plan(self, plan: Plan) -> Dict: diff --git a/src/casdoor/pricing.py b/src/casdoor/pricing.py index 1dd1178..5185d9b 100644 --- a/src/casdoor/pricing.py +++ b/src/casdoor/pricing.py @@ -34,6 +34,28 @@ def __init__(self): self.approveTime = "" self.state = "" + @classmethod + def new(cls, owner, name, created_time, display_name, description, application): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.description = description + self.application = application + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + pricing = cls() + for key, value in data.items(): + if hasattr(pricing, key): + setattr(pricing, key, value) + return pricing + def __str__(self): return str(self.__dict__) @@ -55,7 +77,12 @@ def get_pricings(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - pricings = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + pricings = [] + for pricing in response["data"]: + pricings.append(Pricing.from_dict(pricing)) return pricings def get_pricing(self, pricing_id: str) -> Dict: @@ -72,13 +99,15 @@ def get_pricing(self, pricing_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - pricing = r.json() - return pricing + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + + return Pricing.from_dict(response["data"]) def modify_pricing(self, method: str, pricing: Pricing) -> Dict: url = self.endpoint + f"/api/{method}" - if pricing.owner == "": - pricing.owner = self.org_name + pricing.owner = self.org_name params = { "id": f"{pricing.owner}/{pricing.name}", "clientId": self.client_id, @@ -87,6 +116,8 @@ def modify_pricing(self, method: str, pricing: Pricing) -> Dict: pricing_info = json.dumps(pricing.to_dict()) r = requests.post(url, params=params, data=pricing_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_pricing(self, pricing: Pricing) -> Dict: diff --git a/src/casdoor/product.py b/src/casdoor/product.py index d4ddfb2..8969c9c 100644 --- a/src/casdoor/product.py +++ b/src/casdoor/product.py @@ -39,6 +39,32 @@ def __init__(self): self.state = "" self.providerObjs = [Provider] + @classmethod + def new(cls, owner, name, created_time, display_name, image, description, tag, quantity, sold, state): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.image = image + self.description = description + self.tag = tag + self.quantity = quantity + self.sold = sold + self.state = state + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + product = cls() + for key, value in data.items(): + if hasattr(product, key): + setattr(product, key, value) + return product + def __str__(self): return str(self.__dict__) @@ -60,7 +86,12 @@ def get_products(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - products = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + products = [] + for product in response["data"]: + products.append(Product.from_dict(product)) return products def get_product(self, product_id: str) -> Dict: @@ -77,21 +108,25 @@ def get_product(self, product_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - product = r.json() - return product + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + + return Product.from_dict(response["data"]) def modify_product(self, method: str, product: Product) -> Dict: url = self.endpoint + f"/api/{method}" - if product.owner == "": - product.owner = self.org_name + product.owner = self.org_name params = { "id": f"{product.owner}/{product.name}", "clientId": self.client_id, "clientSecret": self.client_secret, } - product_info = json.dumps(product.to_dict()) + product_info = json.dumps(product.to_dict(), default=self.custom_encoder) r = requests.post(url, params=params, data=product_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_product(self, product: Product) -> Dict: @@ -105,3 +140,7 @@ def update_product(self, product: Product) -> Dict: def delete_product(self, product: Product) -> Dict: response = self.modify_product("delete-product", product) return response + + def custom_encoder(self, o): + if isinstance(o, (Provider)): + return o.__dict__ diff --git a/src/casdoor/provider.py b/src/casdoor/provider.py index 068a9a2..aaff367 100644 --- a/src/casdoor/provider.py +++ b/src/casdoor/provider.py @@ -60,6 +60,27 @@ def __init__(self): self.enableSignAuthnRequest = False self.providerUrl = "" + @classmethod + def new(cls, owner, name, created_time, display_name, category, type): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.category = category + self.type = type + return self + + @classmethod + def from_dict(cls, d: dict): + if not d: + return None + provider = cls() + for key, value in d.items(): + if hasattr(provider, key): + setattr(provider, key, value) + return provider + def __str__(self): return str(self.__dict__) @@ -81,7 +102,12 @@ def get_providers(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - providers = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + providers = [] + for provider in response["data"]: + providers.append(Provider.from_dict(provider)) return providers def get_provider(self, provider_id: str) -> Dict: @@ -98,13 +124,15 @@ def get_provider(self, provider_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - provider = r.json() - return provider + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + + return Provider.from_dict(response["data"]) def modify_provider(self, method: str, provider: Provider) -> Dict: url = self.endpoint + f"/api/{method}" - if provider.owner == "": - provider.owner = self.org_name + provider.owner = self.org_name params = { "id": f"{provider.owner}/{provider.name}", "clientId": self.client_id, @@ -113,6 +141,8 @@ def modify_provider(self, method: str, provider: Provider) -> Dict: provider_info = json.dumps(provider.to_dict()) r = requests.post(url, params=params, data=provider_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_provider(self, provider: Provider) -> Dict: diff --git a/src/casdoor/resource.py b/src/casdoor/resource.py index ab56dc2..696872f 100644 --- a/src/casdoor/resource.py +++ b/src/casdoor/resource.py @@ -35,6 +35,22 @@ def __init__(self): self.url = "" self.description = "" + @classmethod + def new(cls, owner, name): + self = cls() + self.owner = owner + self.name = name + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + resource = cls() + for key, value in data.items(): + setattr(resource, key, value) + return resource + def __str__(self): return str(self.__dict__) @@ -61,7 +77,12 @@ def get_resources(self, owner, user, field, value, sort_field, sort_order) -> Li "clientSecret": self.client_secret, } r = requests.get(url, params) - resources = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + resources = [] + for resource in response["data"]: + resources.append(Resource.from_dict(resource)) return resources def get_resource(self, resource_id: str) -> Dict: @@ -78,13 +99,15 @@ def get_resource(self, resource_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - resource = r.json() - return resource + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + + return Resource.from_dict(response["data"]) def modify_resource(self, method: str, resource: Resource) -> Dict: url = self.endpoint + f"/api/{method}" - if resource.owner == "": - resource.owner = self.org_name + resource.owner = self.org_name params = { "id": f"{resource.owner}/{resource.name}", "clientId": self.client_id, @@ -93,6 +116,8 @@ def modify_resource(self, method: str, resource: Resource) -> Dict: resource_info = json.dumps(resource.to_dict()) r = requests.post(url, params=params, data=resource_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_resource(self, resource: Resource) -> Dict: diff --git a/src/casdoor/role.py b/src/casdoor/role.py index 343b3a4..c2238b1 100644 --- a/src/casdoor/role.py +++ b/src/casdoor/role.py @@ -30,6 +30,27 @@ def __init__(self): self.domains = [""] self.isEnabled = False + @classmethod + def new(cls, owner, name, created_time, display_name, description): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.description = description + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + role = cls() + for key, value in data.items(): + if hasattr(role, key): + setattr(role, key, value) + return role + def __str__(self): return str(self.__dict__) @@ -51,7 +72,12 @@ def get_roles(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - roles = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + roles = [] + for role in response["data"]: + roles.append(Role.from_dict(role)) return roles def get_role(self, role_id: str) -> Dict: @@ -68,13 +94,14 @@ def get_role(self, role_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - role = r.json() - return role + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Role.from_dict(response["data"]) def modify_role(self, method: str, role: Role) -> Dict: url = self.endpoint + f"/api/{method}" - if role.owner == "": - role.owner = self.org_name + role.owner = self.org_name params = { "id": f"{role.owner}/{role.name}", "clientId": self.client_id, @@ -83,6 +110,8 @@ def modify_role(self, method: str, role: Role) -> Dict: role_info = json.dumps(role.to_dict()) r = requests.post(url, params=params, data=role_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_role(self, role: Role) -> Dict: diff --git a/src/casdoor/session.py b/src/casdoor/session.py index bc34985..7a9c287 100644 --- a/src/casdoor/session.py +++ b/src/casdoor/session.py @@ -26,6 +26,26 @@ def __init__(self): self.createdTime = "" self.sessionId = [""] + @classmethod + def new(cls, owner, name, application, created_time, session_id): + self = cls() + self.owner = owner + self.name = name + self.application = application + self.createdTime = created_time + self.sessionId = session_id + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + session = cls() + for key, value in data.items(): + if hasattr(session, key): + setattr(session, key, value) + return session + def __str__(self): return str(self.__dict__) @@ -47,10 +67,15 @@ def get_sessions(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - sessions = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + sessions = [] + for session in response["data"]: + sessions.append(Session.from_dict(session)) return sessions - def get_session(self, session_id: str) -> Dict: + def get_session(self, session_id: str, application: str) -> Dict: """ Get the session from Casdoor providing the session_id. @@ -62,15 +87,17 @@ def get_session(self, session_id: str) -> Dict: "id": f"{self.org_name}/{session_id}", "clientId": self.client_id, "clientSecret": self.client_secret, + "sessionPkId": f"{self.org_name}/{session_id}/{application}", } r = requests.get(url, params) - session = r.json() - return session + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Session.from_dict(response["data"]) def modify_session(self, method: str, session: Session) -> Dict: url = self.endpoint + f"/api/{method}" - if session.owner == "": - session.owner = self.org_name + session.owner = self.org_name params = { "id": f"{session.owner}/{session.name}", "clientId": self.client_id, @@ -79,6 +106,8 @@ def modify_session(self, method: str, session: Session) -> Dict: session_info = json.dumps(session.to_dict()) r = requests.post(url, params=params, data=session_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_session(self, session: Session) -> Dict: diff --git a/src/casdoor/subscription.py b/src/casdoor/subscription.py index e3950dc..de67ec5 100644 --- a/src/casdoor/subscription.py +++ b/src/casdoor/subscription.py @@ -25,8 +25,8 @@ def __init__(self): self.name = "" self.createdTime = "" self.displayName = "" - self.startDate = datetime.now() - self.endDate = datetime.now() + self.startDate = datetime.now().isoformat() + self.endDate = datetime.now().isoformat() self.duration = 0 self.description = "" self.user = "" @@ -37,6 +37,27 @@ def __init__(self): self.approveTime = "" self.state = "" + @classmethod + def new(cls, owner, name, created_time, display_name, description): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.displayName = display_name + self.description = description + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + subscription = cls() + for key, value in data.items(): + if hasattr(subscription, key): + setattr(subscription, key, value) + return subscription + def __str__(self): return str(self.__dict__) @@ -58,7 +79,12 @@ def get_subscriptions(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - subscriptions = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + subscriptions = [] + for subscription in response["data"]: + subscriptions.append(Subscription.from_dict(subscription)) return subscriptions def get_subscription(self, subscription_id: str) -> Dict: @@ -75,13 +101,14 @@ def get_subscription(self, subscription_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - subscription = r.json() - return subscription + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Subscription.from_dict(response["data"]) def modify_subscription(self, method: str, subscription: Subscription) -> Dict: url = self.endpoint + f"/api/{method}" - if subscription.owner == "": - subscription.owner = self.org_name + subscription.owner = self.org_name params = { "id": f"{subscription.owner}/{subscription.name}", "clientId": self.client_id, @@ -90,6 +117,8 @@ def modify_subscription(self, method: str, subscription: Subscription) -> Dict: subscription_info = json.dumps(subscription.to_dict()) r = requests.post(url, params=params, data=subscription_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_subscription(self, subscription: Subscription) -> Dict: diff --git a/src/casdoor/syncer.py b/src/casdoor/syncer.py index e9e6c9d..469ce38 100644 --- a/src/casdoor/syncer.py +++ b/src/casdoor/syncer.py @@ -57,6 +57,48 @@ def __init__(self): self.isReadOnly = False self.isEnabled = False + @classmethod + def new( + cls, + owner, + name, + created_time, + organization, + host, + port, + user, + password, + database_type, + database, + table, + sync_interval, + ): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.organization = organization + self.host = host + self.port = port + self.user = user + self.password = password + self.databaseType = database_type + self.database = database + self.table = table + self.syncInterval = sync_interval + return self + + @classmethod + def from_dict(cls, d: dict): + if not d: + return None + + syncer = cls() + for key, value in d.items(): + if hasattr(syncer, key): + setattr(syncer, key, value) + return syncer + def __str__(self): return str(self.__dict__) @@ -78,7 +120,12 @@ def get_syncers(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - syncers = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + syncers = [] + for syncer in response["data"]: + syncers.append(Syncer.from_dict(syncer)) return syncers def get_syncer(self, syncer_id: str) -> Dict: @@ -95,21 +142,24 @@ def get_syncer(self, syncer_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - syncer = r.json() - return syncer + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Syncer.from_dict(response["data"]) def modify_syncer(self, method: str, syncer: Syncer) -> Dict: url = self.endpoint + f"/api/{method}" - if syncer.owner == "": - syncer.owner = self.org_name + syncer.owner = self.org_name params = { "id": f"{syncer.owner}/{syncer.name}", "clientId": self.client_id, "clientSecret": self.client_secret, } - syncer_info = json.dumps(syncer.to_dict()) + syncer_info = json.dumps(syncer.to_dict(), default=self.custom_encoder) r = requests.post(url, params=params, data=syncer_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_syncer(self, syncer: Syncer) -> Dict: @@ -123,3 +173,7 @@ def update_syncer(self, syncer: Syncer) -> Dict: def delete_syncer(self, syncer: Syncer) -> Dict: response = self.modify_syncer("delete-syncer", syncer) return response + + def custom_encoder(self, o): + if isinstance(o, (TableColumn)): + return o.__dict__ diff --git a/src/casdoor/token.py b/src/casdoor/token.py index 1369dd2..076c4e7 100644 --- a/src/casdoor/token.py +++ b/src/casdoor/token.py @@ -36,6 +36,54 @@ def __init__(self): self.codeIsUsed = False self.codeExpireIn = 0 + @classmethod + def new( + cls, + owner, + name, + created_time, + application, + organization, + user, + code, + access_token, + refresh_token, + expires_in, + scope, + token_type, + code_challenge, + code_is_used, + code_expire_in, + ): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.application = application + self.organization = organization + self.user = user + self.code = code + self.accessToken = access_token + self.refreshToken = refresh_token + self.expiresIn = expires_in + self.scope = scope + self.tokenType = token_type + self.codeChallenge = code_challenge + self.codeIsUsed = code_is_used + self.codeExpireIn = code_expire_in + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + token = cls() + for key, value in data.items(): + if hasattr(token, key): + setattr(token, key, value) + return token + def __str__(self): return str(self.__dict__) @@ -59,7 +107,12 @@ def get_tokens(self, p, page_size) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - tokens = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + tokens = [] + for token in response["data"]: + tokens.append(Token.from_dict(token)) return tokens def get_token(self, token_id: str) -> Dict: @@ -76,8 +129,10 @@ def get_token(self, token_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - token = r.json() - return token + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Token.from_dict(response["data"]) def modify_token(self, method: str, token: Token) -> Dict: url = self.endpoint + f"/api/{method}" @@ -91,6 +146,8 @@ def modify_token(self, method: str, token: Token) -> Dict: token_info = json.dumps(token.to_dict()) r = requests.post(url, params=params, data=token_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_token(self, token: Token) -> Dict: diff --git a/src/casdoor/user.py b/src/casdoor/user.py index 40ba41b..89f3d68 100644 --- a/src/casdoor/user.py +++ b/src/casdoor/user.py @@ -51,6 +51,26 @@ def __init__(self): self.wechat = "" self.weibo = "" + @classmethod + def new(cls, owner, name, created_time, display_name): + self = cls() + self.name = name + self.owner = owner + self.createdTime = created_time + self.displayName = display_name + return self + + @classmethod + def from_dict(cls, data: dict): + if data is None: + return None + + user = cls() + for key, value in data.items(): + if hasattr(user, key): + setattr(user, key, value) + return user + def __str__(self): return str(self.__dict__) @@ -72,7 +92,12 @@ def get_users(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - users = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + users = [] + for user in response["data"]: + users.append(User.from_dict(user)) return users def get_user(self, user_id: str) -> Dict: @@ -89,8 +114,10 @@ def get_user(self, user_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - user = r.json() - return user + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return User.from_dict(response["data"]) def get_user_count(self, is_online: bool = None) -> int: """ @@ -117,8 +144,7 @@ def get_user_count(self, is_online: bool = None) -> int: def modify_user(self, method: str, user: User) -> Dict: url = self.endpoint + f"/api/{method}" - if user.owner == "": - user.owner = self.org_name + user.owner = self.org_name params = { "id": f"{user.owner}/{user.name}", "clientId": self.client_id, @@ -127,6 +153,8 @@ def modify_user(self, method: str, user: User) -> Dict: user_info = json.dumps(user.to_dict()) r = requests.post(url, params=params, data=user_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_user(self, user: User) -> Dict: diff --git a/src/casdoor/webhook.py b/src/casdoor/webhook.py index 4f40308..dbd49b3 100644 --- a/src/casdoor/webhook.py +++ b/src/casdoor/webhook.py @@ -43,6 +43,25 @@ def __init__(self): self.isReadOnly = False self.isEnabled = False + @classmethod + def new(cls, owner, name, created_time, organization): + self = cls() + self.owner = owner + self.name = name + self.createdTime = created_time + self.organization = organization + return self + + @classmethod + def from_dict(cls, d: dict): + if d is None: + return None + webhook = cls() + for key, value in d.items(): + if hasattr(webhook, key): + setattr(webhook, key, value) + return webhook + def __str__(self): return str(self.__dict__) @@ -64,7 +83,12 @@ def get_webhooks(self) -> List[Dict]: "clientSecret": self.client_secret, } r = requests.get(url, params) - webhooks = r.json() + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + webhooks = [] + for webhook in response["data"]: + webhooks.append(Webhook.from_dict(webhook)) return webhooks def get_webhook(self, webhook_id: str) -> Dict: @@ -81,21 +105,24 @@ def get_webhook(self, webhook_id: str) -> Dict: "clientSecret": self.client_secret, } r = requests.get(url, params) - webhook = r.json() - return webhook + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return Webhook.from_dict(response["data"]) def modify_webhook(self, method: str, webhook: Webhook) -> Dict: url = self.endpoint + f"/api/{method}" - if webhook.owner == "": - webhook.owner = self.org_name + webhook.owner = self.org_name params = { "id": f"{webhook.owner}/{webhook.name}", "clientId": self.client_id, "clientSecret": self.client_secret, } - webhook_info = json.dumps(webhook.to_dict()) + webhook_info = json.dumps(webhook.to_dict(), default=self.custom_encoder) r = requests.post(url, params=params, data=webhook_info) response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) return response def add_webhook(self, webhook: Webhook) -> Dict: @@ -109,3 +136,7 @@ def update_webhook(self, webhook: Webhook) -> Dict: def delete_webhook(self, webhook: Webhook) -> Dict: response = self.modify_webhook("delete-webhook", webhook) return response + + def custom_encoder(self, o): + if isinstance(o, (TableColumn)): + return o.__dict__ diff --git a/src/tests/test_adapter.py b/src/tests/test_adapter.py new file mode 100644 index 0000000..3c63ba5 --- /dev/null +++ b/src/tests/test_adapter.py @@ -0,0 +1,97 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.adapter import Adapter +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class AdapterTest(unittest.TestCase): + def test_adapter(self): + name = get_random_name("Adapter") + + # Add a new object + adapter = Adapter.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + host=name, + user="https://casdoor.org", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + + try: + sdk.add_adapter(adapter=adapter) + except Exception as e: + self.fail("Failed to add object: " + str(e)) + + # Get all objects, check if our added object is inside the list + try: + adapters = sdk.get_adapters() + except Exception as e: + self.fail("Failed to get objects: " + str(e)) + names = [item.name for item in adapters] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + adapter = sdk.get_adapter(name) + except Exception as e: + self.fail("Failed to get object: " + str(e)) + + self.assertEqual(adapter.name, name, "Retrieved object does not match added object") + + # Update the object + updated_user = "Updated Casdoor Website" + adapter.user = updated_user + try: + sdk.update_adapter(adapter) + except Exception as e: + self.fail("Failed to update object: " + str(e)) + + # Validate the update + try: + updated_adapter = sdk.get_adapter(name) + except Exception as e: + self.fail("Failed to get updated object: " + str(e)) + + self.assertEqual(updated_adapter.user, updated_user, "Failed to update object, display_name mismatch") + + # Delete the object + try: + sdk.delete_adapter(adapter) + except Exception as e: + self.fail("Failed to delete object: " + str(e)) + + # Validate the deletion + try: + deleted_adapter = sdk.get_adapter(name) + except Exception as e: + self.fail("Failed to delete object: " + str(e)) + + self.assertIsNone(deleted_adapter, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_cert.py b/src/tests/test_cert.py new file mode 100644 index 0000000..8c95379 --- /dev/null +++ b/src/tests/test_cert.py @@ -0,0 +1,103 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.cert import Cert +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class CertTest(unittest.TestCase): + def test_cert(self): + name = get_random_name("Cert") + + # Add a new object + cert = Cert.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + scope="JWT", + type="x509", + crypto_algorithm="RS256", + bit_size=4096, + expire_in_years=20, + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + + try: + sdk.add_cert(cert=cert) + except Exception as e: + self.fail("Failed to add object: " + str(e)) + + # Get all objects, check if our added object is inside the list + try: + certs = sdk.get_certs() + except Exception as e: + self.fail("Failed to get objects: " + str(e)) + names = [item.name for item in certs] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + cert = sdk.get_cert(name) + except Exception as e: + self.fail("Failed to get object: " + str(e)) + + self.assertEqual(cert.name, name, "Retrieved object does not match added object") + + # Update the object + updated_display_name = "Updated Casdoor Website" + cert.displayName = updated_display_name + try: + sdk.update_cert(cert) + except Exception as e: + self.fail("Failed to update object: " + str(e)) + + # Validate the update + try: + updated_cert = sdk.get_cert(name) + except Exception as e: + self.fail("Failed to get updated object: " + str(e)) + + self.assertEqual( + updated_cert.displayName, updated_display_name, "Failed to update object, display_name mismatch" + ) + + # Delete the object + try: + sdk.delete_cert(cert) + except Exception as e: + self.fail("Failed to delete object: " + str(e)) + + # Validate the deletion + try: + deleted_cert = sdk.get_cert(name) + except Exception as e: + self.fail("Failed to delete object: " + str(e)) + + self.assertIsNone(deleted_cert, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_enforcer.py b/src/tests/test_enforcer.py new file mode 100644 index 0000000..c5ebc03 --- /dev/null +++ b/src/tests/test_enforcer.py @@ -0,0 +1,95 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.enforcer import Enforcer +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class EnforcerTest(unittest.TestCase): + def test_enforcer(self): + name = get_random_name("Enforcer") + + # Add a new object + enforcer = Enforcer.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + description="built-in/user-model-built-in", + model="built-in/user-adapter-built-in", + adapter="Casdoor Website", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_enforcer(enforcer=enforcer) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + enforcers = sdk.get_enforcers() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in enforcers] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + enforcer = sdk.get_enforcer(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(enforcer.name, name) + + # Update the object + updated_description = "Updated Casdoor Website" + enforcer.description = updated_description + try: + sdk.update_enforcer(enforcer) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_enforcer = sdk.get_enforcer(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_enforcer.description, updated_description) + + # Delete the object + try: + sdk.delete_enforcer(enforcer) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_enforcer = sdk.get_enforcer(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_enforcer, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_model.py b/src/tests/test_model.py new file mode 100644 index 0000000..25b57b9 --- /dev/null +++ b/src/tests/test_model.py @@ -0,0 +1,106 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.model import Model +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class ModelTest(unittest.TestCase): + def test_model(self): + name = get_random_name("model") + + # Add a new object + model = Model.new( + owner="casbin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + model_text="[request_definition]\n" + + "r = sub, obj, act\n" + + "\n" + + "[policy_definition]\n" + + "p = sub, obj, act\n" + + "\n" + + "[role_definition]\n" + + "g = _, _\n" + + "\n" + + "[policy_effect]\n" + + "e = some(where (p.eft == allow))\n" + + "\n" + + "[matchers]\n" + + "m = g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_model(model=model) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + models = sdk.get_models() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in models] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + model = sdk.get_model(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(model.name, name) + + # Update the object + updated_display_name = "Updated Casdoor Website" + model.displayName = updated_display_name + try: + sdk.update_model(model) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_model = sdk.get_model(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_model.displayName, updated_display_name) + + # Delete the object + try: + sdk.delete_model(model) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_model = sdk.get_model(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_model, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_payment.py b/src/tests/test_payment.py new file mode 100644 index 0000000..f926c1d --- /dev/null +++ b/src/tests/test_payment.py @@ -0,0 +1,93 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.payment import Payment +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class PaymentTest(unittest.TestCase): + def test_payment(self): + name = get_random_name("Payment") + + # Add a new object + payment = Payment.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + product_name="casbin", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_payment(payment=payment) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + payments = sdk.get_payments() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in payments] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + payment = sdk.get_payment(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(payment.name, name) + + # Update the object + updated_product_name = "Updated Casdoor Website" + payment.productName = updated_product_name + try: + sdk.update_payment(payment) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_payment = sdk.get_payment(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_payment.productName, updated_product_name) + + # Delete the object + try: + sdk.delete_payment(payment) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_payment = sdk.get_payment(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_payment, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_permission.py b/src/tests/test_permission.py new file mode 100644 index 0000000..da45892 --- /dev/null +++ b/src/tests/test_permission.py @@ -0,0 +1,102 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.permission import Permission +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class permissionTest(unittest.TestCase): + def test_permission(self): + name = get_random_name("Permission") + + # Add a new object + permission = Permission.new( + owner="casbin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + description="Casdoor Website", + users=["casbin/*"], + roles=[], + domains=[], + model="user-model-built-in", + resource_type="Application", + resources=["app-casbin"], + actions=["Read", "Write"], + effect="Allow", + is_enabled=True, + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_permission(permission=permission) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + permissions = sdk.get_permissions() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in permissions] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + permission = sdk.get_permission(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(permission.name, name) + + # Update the object + updated_description = "Updated Casdoor Website" + permission.description = updated_description + try: + sdk.update_permission(permission) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_permission = sdk.get_permission(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_permission.description, updated_description) + + # Delete the object + try: + sdk.delete_permission(permission) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_permission = sdk.get_permission(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_permission, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_plan.py b/src/tests/test_plan.py new file mode 100644 index 0000000..1e2b11b --- /dev/null +++ b/src/tests/test_plan.py @@ -0,0 +1,93 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.plan import Plan +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class PlanTest(unittest.TestCase): + def test_plan(self): + name = get_random_name("Plan") + + # Add a new object + plan = Plan.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + description="casbin", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_plan(plan=plan) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + plans = sdk.get_plans() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in plans] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + plan = sdk.get_plan(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(plan.name, name) + + # Update the object + updated_description = "Updated Casdoor Website" + plan.description = updated_description + try: + sdk.update_plan(plan) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_plan = sdk.get_plan(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_plan.description, updated_description) + + # Delete the object + try: + sdk.delete_plan(plan) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_plan = sdk.get_plan(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_plan, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_pricing.py b/src/tests/test_pricing.py new file mode 100644 index 0000000..e7af867 --- /dev/null +++ b/src/tests/test_pricing.py @@ -0,0 +1,94 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.pricing import Pricing +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class pricingTest(unittest.TestCase): + def test_pricing(self): + name = get_random_name("Pricing") + + # Add a new object + pricing = Pricing.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + description="app-admin", + application="Casdoor Website", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_pricing(pricing=pricing) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + pricings = sdk.get_pricings() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in pricings] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + pricing = sdk.get_pricing(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(pricing.name, name) + + # Update the object + updated_description = "Updated Casdoor Website" + pricing.description = updated_description + try: + sdk.update_pricing(pricing) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_pricing = sdk.get_pricing(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_pricing.description, updated_description) + + # Delete the object + try: + sdk.delete_pricing(pricing) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_pricing = sdk.get_pricing(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_pricing, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_product.py b/src/tests/test_product.py new file mode 100644 index 0000000..135437a --- /dev/null +++ b/src/tests/test_product.py @@ -0,0 +1,98 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.product import Product +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class ProductTest(unittest.TestCase): + def test_product(self): + name = get_random_name("Product") + + # Add a new object + product = Product.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + image="https://cdn.casbin.org/img/casdoor-logo_1185x256.png", + description="Casdoor Website", + tag="auto_created_product_for_plan", + quantity=999, + sold=0, + state="Published", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_product(product=product) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + products = sdk.get_products() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in products] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + product = sdk.get_product(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(product.name, name) + + # Update the object + updated_description = "Updated Casdoor Website" + product.description = updated_description + try: + sdk.update_product(product) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_product = sdk.get_product(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_product.description, updated_description) + + # Delete the object + try: + sdk.delete_product(product) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_product = sdk.get_product(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_product, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_provider.py b/src/tests/test_provider.py new file mode 100644 index 0000000..b5ae7db --- /dev/null +++ b/src/tests/test_provider.py @@ -0,0 +1,94 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.provider import Provider +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class ProviderTest(unittest.TestCase): + def test_provider(self): + name = get_random_name("Provider") + + # Add a new object + provider = Provider.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + category="Captcha", + type="Default", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_provider(provider=provider) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + providers = sdk.get_providers() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in providers] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + provider = sdk.get_provider(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(provider.name, name) + + # Update the object + updated_display_name = "Updated Casdoor Website" + provider.displayName = updated_display_name + try: + sdk.update_provider(provider) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_provider = sdk.get_provider(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_provider.displayName, updated_display_name) + + # Delete the object + try: + sdk.delete_provider(provider) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_provider = sdk.get_provider(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_provider, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_role.py b/src/tests/test_role.py new file mode 100644 index 0000000..0e839ee --- /dev/null +++ b/src/tests/test_role.py @@ -0,0 +1,93 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.role import Role +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class RoleTest(unittest.TestCase): + def test_role(self): + name = get_random_name("Role") + + # Add a new object + role = Role.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + description="Casdoor Website", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_role(role=role) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + roles = sdk.get_roles() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in roles] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + role = sdk.get_role(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(role.name, name) + + # Update the object + updated_description = "Updated Casdoor Website" + role.description = updated_description + try: + sdk.update_role(role) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_role = sdk.get_role(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_role.description, updated_description) + + # Delete the object + try: + sdk.delete_role(role) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_role = sdk.get_role(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_role, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_session.py b/src/tests/test_session.py new file mode 100644 index 0000000..ea90db4 --- /dev/null +++ b/src/tests/test_session.py @@ -0,0 +1,93 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.session import Session +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class sessionTest(unittest.TestCase): + def test_session(self): + name = get_random_name("Session") + + # Add a new object + session = Session.new( + owner="casbin", + name=name, + application="app-built-in", + created_time=datetime.datetime.now().isoformat(), + session_id=[], + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_session(session=session) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + sessions = sdk.get_sessions() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in sessions] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + session = sdk.get_session(name, session.application) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(session.name, name) + + # Update the object + updated_time = "Updated Casdoor Website" + session.createdTime = updated_time + try: + sdk.update_session(session) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_session = sdk.get_session(name, session.application) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_session.createdTime, updated_time) + + # Delete the object + try: + sdk.delete_session(session) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_session = sdk.get_session(name, session.application) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_session, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_subscription.py b/src/tests/test_subscription.py new file mode 100644 index 0000000..3d9111b --- /dev/null +++ b/src/tests/test_subscription.py @@ -0,0 +1,93 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.subscription import Subscription +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class SubscriptionTest(unittest.TestCase): + def test_subscription(self): + name = get_random_name("Subscription") + + # Add a new object + subscription = Subscription.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + description="Casdoor Website", + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_subscription(subscription=subscription) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + subscriptions = sdk.get_subscriptions() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in subscriptions] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + subscription = sdk.get_subscription(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(subscription.name, name) + + # Update the object + updated_description = "Updated Casdoor Website" + subscription.description = updated_description + try: + sdk.update_subscription(subscription) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_subscription = sdk.get_subscription(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_subscription.description, updated_description) + + # Delete the object + try: + sdk.delete_subscription(subscription) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_subscription = sdk.get_subscription(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_subscription, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_syncer.py b/src/tests/test_syncer.py new file mode 100644 index 0000000..a53317e --- /dev/null +++ b/src/tests/test_syncer.py @@ -0,0 +1,100 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.syncer import Syncer +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class SyncerTest(unittest.TestCase): + def test_syncer(self): + name = get_random_name("Syncer") + + # Add a new object + syncer = Syncer.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + organization="casbin", + host="localhost", + port=3306, + user="root", + password="123", + database_type="mysql", + database="syncer_db", + table="user-table", + sync_interval=1, + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_syncer(syncer=syncer) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + syncers = sdk.get_syncers() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in syncers] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + syncer = sdk.get_syncer(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(syncer.name, name) + + # Update the object + updated_password = "123456" + syncer.password = updated_password + try: + sdk.update_syncer(syncer) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_syncer = sdk.get_syncer(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_syncer.password, updated_password) + + # Delete the object + try: + sdk.delete_syncer(syncer) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_syncer = sdk.get_syncer(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_syncer, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_user.py b/src/tests/test_user.py new file mode 100644 index 0000000..2d599f1 --- /dev/null +++ b/src/tests/test_user.py @@ -0,0 +1,87 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.user import User +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class UserTest(unittest.TestCase): + def test_user(self): + name = get_random_name("User") + + # Add a new object + user = User.new(owner="admin", name=name, created_time=datetime.datetime.now().isoformat(), display_name=name) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_user(user=user) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + users = sdk.get_users() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in users] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + user = sdk.get_user(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(user.name, name) + + # Update the object + updated_display_name = "Updated Casdoor Website" + user.displayName = updated_display_name + try: + sdk.update_user(user) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_user = sdk.get_user(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_user.displayName, updated_display_name) + + # Delete the object + try: + sdk.delete_user(user) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_user = sdk.get_user(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_user, "Failed to delete object, it's still retrievable") diff --git a/src/tests/test_webhook.py b/src/tests/test_webhook.py new file mode 100644 index 0000000..e6cee03 --- /dev/null +++ b/src/tests/test_webhook.py @@ -0,0 +1,89 @@ +# Copyright 2023 The Casdoor Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import datetime +import unittest + +from src.casdoor import CasdoorSDK +from src.casdoor.webhook import Webhook +from src.tests.test_util import ( + TestApplication, + TestClientId, + TestClientSecret, + TestEndpoint, + TestJwtPublicKey, + TestOrganization, + get_random_name, +) + + +class WebhookTest(unittest.TestCase): + def test_webhook(self): + name = get_random_name("Webhook") + + # Add a new object + webhook = Webhook.new( + owner="casbin", name=name, created_time=datetime.datetime.now().isoformat(), organization="casbin" + ) + + sdk = CasdoorSDK( + TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + ) + try: + sdk.add_webhook(webhook=webhook) + except Exception as e: + self.fail(f"Failed to add object: {e}") + + # Get all objects, check if our added object is inside the list + try: + webhooks = sdk.get_webhooks() + except Exception as e: + self.fail(f"Failed to get objects: {e}") + names = [item.name for item in webhooks] + self.assertIn(name, names, "Added object not found in list") + + # Get the object + try: + webhook = sdk.get_webhook(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertEqual(webhook.name, name) + + # Update the object + updated_organization = "Updated Casdoor Website" + webhook.organization = updated_organization + try: + sdk.update_webhook(webhook) + except Exception as e: + self.fail(f"Failed to update object: {e}") + + # Validate the update + try: + updated_webhook = sdk.get_webhook(name) + except Exception as e: + self.fail(f"Failed to get updated object: {e}") + self.assertEqual(updated_webhook.organization, updated_organization) + + # Delete the object + try: + sdk.delete_webhook(webhook) + except Exception as e: + self.fail(f"Failed to delete object: {e}") + + # Validate the deletion + try: + deleted_webhook = sdk.get_webhook(name) + except Exception as e: + self.fail(f"Failed to get object: {e}") + self.assertIsNone(deleted_webhook, "Failed to delete object, it's still retrievable")