Compare commits

..

11 Commits

Author SHA1 Message Date
Lebaud Antoine 64150da81a wip 2024-02-15 19:17:30 +01:00
Lebaud Antoine 6c0acd0596 ✏️(project) fix typos
Found typos, fixed them.
2024-02-15 16:48:00 +01:00
Lebaud Antoine 625e98515b 🔧(backend) activate container liveness probes
Enabled Dockerflow Django app by activating liveness probes. The previously
unavailable routes such as `__heartbeat__` and `__lbheartbeat__` are now
accessible. New endpoints include:
* GET /__version__
* GET /__heartbeat__
* GET /__lbheartbeat__
2024-02-15 16:48:00 +01:00
Lebaud Antoine 297007162a (frontend) introduce frontend Docker Image
To facilitate deployment on Kubernetes, we've introduced a Docker image for the
frontend. The Next project is built, and its static output is served using an
Nginx reverse proxy.

Since DevOps lacks a certified cold storage solution (e.g., S3) for serving
static files, we've opted to containerize the frontend as a quick workaround for
deploying staging environments.
2024-02-15 16:48:00 +01:00
Lebaud Antoine 8fa947f24d ⬆️(project) upgrade base Docker image for mail-builder
Updated to Node Image version 20 to align with the upcoming frontend image.
2024-02-15 16:48:00 +01:00
Lebaud Antoine ee580ad258 🐛(project) run production image locally with docker-compose
The local deployment of the Production image through docker-compose was
failing due to issues in the Django configurations, influenced by Joanie.

The bug stemmed from a dependency on a development-specific package
(drf-spectacular-sidecar) while attempting to run the application in
production mode.

Changes Made:
- Introduced new Django settings for local demo environments.
- Uncommented the nginx configuration to address the production image
  deployment issues.
2024-02-15 16:48:00 +01:00
Lebaud Antoine 155043e0eb wip replace Implicit flow by Authorization Code flow in frontend
Instead of interacting with Keycloak, the frontend navigate to the
/authenticate endpoint, which starts the Authorization code flow.

When the flow is done, the backend redirect back to the SPA,
passing a session cookie and a csrf cookie.

Done:
* Query GET user/me to determine if user is authenticated yet
* Remove Keycloak js dependency, as all the OIDC logic is handled by the backend
* Store user's data instead of the JWT token

Todo in a dedicated PR:
* Implement the home screen with the SSO login buton
* Handle Logout properly
* Intercept 401 and 403 error properly
2024-02-15 16:46:32 +01:00
Lebaud Antoine df03534fb1 wip handle Authorization code flow with Django
Integrate 'mozilla-django-oidc' dependency, to support
Authorization Code flow, which is required by Agent Connect.

Thus, we provide a secure back channel OIDC flow, and return
to the client only a session cookie.

Done:
* Replace JWT authentication by Session based authentication in DRF
* Update Django settings to make OIDC configurations easily editable
* Add 'mozilla-django-oidc' routes to our router
* Implement a custom Django Authentication class to adapt 'mozilla-django-oidc' to our needs

'mozilla-django-oidc' routes added are:
* /authenticate
* /callback (the redirect_uri called back by the Idp)
* /logout

Todo in dedicated PR:
* Configure redis to manage these session cookies
* Test and adjust code if necessary when the session cookie expires
* Test and adjust code if necessary when the user logout
2024-02-15 16:46:16 +01:00
Lebaud Antoine 714d8188f5 wip drop JWT authentication in API tests
force login to bypass authorization checks when necessary.

Note: Generating a session cookie through OIDC flow
is not supported while testing our API.
2024-02-15 12:03:05 +01:00
Lebaud Antoine 65b36c6116 wip proxy keycloak with nginx
Backend and Frontend send requests to Keycloak through Nginx.

Thus, all requests from frontend and backend shared a same host
when received by Keycloak.

Otherwise, the flow is initiated from http://localhost:8080. When the Backend
calls token endpoint from Keycloak container at http://keycloak:8080,
the JWT token issuer and sender are mismatching.
2024-02-15 12:02:53 +01:00
Lebaud Antoine 52d8969602 wip update keycloak config
* rename client people-front to people
* add a client secret shared with the backend
* add allowed redirect uris
* disable implicit flow and enable Authorization Code flow without PCKE
* sign userinfo endpoint to return application/jwt content
2024-02-14 22:35:15 +01:00
349 changed files with 7611 additions and 20057 deletions
-3
View File
@@ -31,6 +31,3 @@ db.sqlite3
.mypy_cache
.pylint.d
.pytest_cache
# Frontend dependencies
node_modules
-51
View File
@@ -1,51 +0,0 @@
name: Deploy
on:
push:
tags:
- 'preprod'
jobs:
notify-argocd:
runs-on: ubuntu-latest
steps:
-
uses: actions/create-github-app-token@v1
id: app-token
with:
app-id: ${{ secrets.APP_ID }}
private-key: ${{ secrets.PRIVATE_KEY }}
owner: ${{ github.repository_owner }}
repositories: "people,secrets"
-
name: Checkout repository
uses: actions/checkout@v2
with:
submodules: recursive
token: ${{ steps.app-token.outputs.token }}
-
name: Load sops secrets
uses: rouja/actions-sops@main
with:
secret-file: secrets/numerique-gouv/people/secrets.enc.env
age-key: ${{ secrets.SOPS_PRIVATE }}
-
name: Call argocd github webhook
run: |
data='{"ref": "'$GITHUB_REF'","repository": {"html_url":"'$GITHUB_SERVER_URL'/'$GITHUB_REPOSITORY'"}}'
sig=$(echo -n ${data} | openssl dgst -sha1 -hmac ''${ARGOCD_WEBHOOK_SECRET}'' | awk '{print "X-Hub-Signature: sha1="$2}')
curl -X POST -H 'X-GitHub-Event:push' -H "Content-Type: application/json" -H "${sig}" --data "${data}" $ARGOCD_WEBHOOK_URL
sig=$(echo -n ${data} | openssl dgst -sha1 -hmac ''${ARGOCD_PRODUCTION_WEBHOOK_SECRET}'' | awk '{print "X-Hub-Signature: sha1="$2}')
curl -X POST -H 'X-GitHub-Event:push' -H "Content-Type: application/json" -H "${sig}" --data "${data}" $ARGOCD_PRODUCTION_WEBHOOK_URL
start-test-on-preprod:
needs:
- notify-argocd
runs-on: ubuntu-latest
if: startsWith(github.event.ref, 'refs/tags/preprod')
steps:
-
name: Debug
run: |
echo "Start test when preprod is ready"
+13 -98
View File
@@ -5,135 +5,50 @@ on:
push:
branches:
- 'main'
- 'fix/warning-backend'
tags:
- 'v*'
pull_request:
branches:
- 'main'
env:
DOCKER_USER: 1001:127
jobs:
build-and-push-backend:
build-and-push:
runs-on: ubuntu-latest
steps:
-
uses: actions/create-github-app-token@v1
id: app-token
with:
app-id: ${{ secrets.APP_ID }}
private-key: ${{ secrets.PRIVATE_KEY }}
owner: ${{ github.repository_owner }}
repositories: "people,secrets"
-
name: Checkout repository
uses: actions/checkout@v2
with:
submodules: recursive
token: ${{ steps.app-token.outputs.token }}
name: Checkout
uses: actions/checkout@v4
-
name: Docker meta
id: meta
uses: docker/metadata-action@v5
with:
images: lasuite/people-backend
images: lasuite/people
-
name: Load sops secrets
uses: rouja/actions-sops@main
with:
secret-file: secrets/numerique-gouv/people/secrets.enc.env
secret-file: .github/workflows/secrets.enc.env
age-key: ${{ secrets.SOPS_PRIVATE }}
-
name: Login to DockerHub
if: github.event_name != 'pull_request'
run: echo "$DOCKER_HUB_PASSWORD" | docker login -u "$DOCKER_HUB_USER" --password-stdin
-
name: Build and push
name: Build and push backend
uses: docker/build-push-action@v5
with:
context: .
target: backend-production
build-args: DOCKER_USER=${{ env.DOCKER_USER }}:-1000
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta.outputs.tags }}
target: production
push: true
tags: app-${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
build-and-push-frontend:
runs-on: ubuntu-latest
steps:
-
uses: actions/create-github-app-token@v1
id: app-token
with:
app-id: ${{ secrets.APP_ID }}
private-key: ${{ secrets.PRIVATE_KEY }}
owner: ${{ github.repository_owner }}
repositories: "people,secrets"
-
name: Checkout repository
uses: actions/checkout@v2
with:
submodules: recursive
token: ${{ steps.app-token.outputs.token }}
-
name: Docker meta
id: meta
uses: docker/metadata-action@v5
with:
images: lasuite/people-frontend
-
name: Load sops secrets
uses: rouja/actions-sops@main
with:
secret-file: secrets/numerique-gouv/people/secrets.enc.env
age-key: ${{ secrets.SOPS_PRIVATE }}
-
name: Login to DockerHub
if: github.event_name != 'pull_request'
run: echo "$DOCKER_HUB_PASSWORD" | docker login -u "$DOCKER_HUB_USER" --password-stdin
-
name: Build and push
name: Build and push frontend
uses: docker/build-push-action@v5
with:
context: .
target: frontend-production
build-args: DOCKER_USER=${{ env.DOCKER_USER }}:-1000
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta.outputs.tags }}
target: frontend
push: true
tags: frontend-${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
notify-argocd:
needs:
- build-and-push-frontend
- build-and-push-backend
runs-on: ubuntu-latest
if: github.event_name != 'pull_request'
steps:
-
uses: actions/create-github-app-token@v1
id: app-token
with:
app-id: ${{ secrets.APP_ID }}
private-key: ${{ secrets.PRIVATE_KEY }}
owner: ${{ github.repository_owner }}
repositories: "people,secrets"
-
name: Checkout repository
uses: actions/checkout@v2
with:
submodules: recursive
token: ${{ steps.app-token.outputs.token }}
-
name: Load sops secrets
uses: rouja/actions-sops@main
with:
secret-file: secrets/numerique-gouv/people/secrets.enc.env
age-key: ${{ secrets.SOPS_PRIVATE }}
-
name: Call argocd github webhook
run: |
data='{"ref": "'$GITHUB_REF'","repository": {"html_url":"'$GITHUB_SERVER_URL'/'$GITHUB_REPOSITORY'"}}'
sig=$(echo -n ${data} | openssl dgst -sha1 -hmac ''${ARGOCD_WEBHOOK_SECRET}'' | awk '{print "X-Hub-Signature: sha1="$2}')
curl -X POST -H 'X-GitHub-Event:push' -H "Content-Type: application/json" -H "${sig}" --data "${data}" $ARGOCD_WEBHOOK_URL
+53 -127
View File
@@ -4,6 +4,8 @@ on:
push:
branches:
- main
tags:
- 'v*'
pull_request:
branches:
- '*'
@@ -14,7 +16,7 @@ jobs:
if: github.event_name == 'pull_request' # Makes sense only for pull requests
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
with:
fetch-depth: 0
- name: show
@@ -37,7 +39,7 @@ jobs:
github.event_name == 'pull_request'
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
with:
fetch-depth: 0
- name: Check that the CHANGELOG has been modified in the current branch
@@ -47,7 +49,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
- name: Check CHANGELOG max line length
run: |
max_line_length=$(cat CHANGELOG.md | grep -Ev "^\[.*\]: https://github.com" | wc -L)
@@ -56,9 +58,11 @@ jobs:
exit 1
fi
install-front:
test-front:
runs-on: ubuntu-latest
defaults:
run:
working-directory: src/frontend/
steps:
- name: Checkout repository
uses: actions/checkout@v4
@@ -67,85 +71,42 @@ jobs:
uses: actions/setup-node@v4
with:
node-version: '18.x'
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
cache: 'yarn'
cache-dependency-path: src/frontend/yarn.lock
- name: Install dependencies
if: steps.front-node_modules.outputs.cache-hit != 'true'
run: cd src/frontend/ && yarn install --frozen-lockfile
- name: Cache install frontend
if: steps.front-node_modules.outputs.cache-hit != 'true'
uses: actions/cache@v4
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
build-front:
runs-on: ubuntu-latest
needs: install-front
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
- name: Build CI App
run: cd src/frontend/ && yarn ci:build
- name: Cache build frontend
uses: actions/cache@v4
with:
path: src/frontend/apps/desk/out/
key: build-front-${{ github.run_id }}
test-front:
runs-on: ubuntu-latest
needs: install-front
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
run: yarn install --frozen-lockfile
- name: Test App
run: cd src/frontend/ && yarn app:test
run: yarn app:test
- name: Test Translations
run: yarn i18n:test
lint-front:
runs-on: ubuntu-latest
needs: install-front
defaults:
run:
working-directory: src/frontend/
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
- name: Setup Node.js
uses: actions/setup-node@v4
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
node-version: '18.x'
cache: 'yarn'
cache-dependency-path: src/frontend/yarn.lock
- name: Install dependencies
run: yarn install --frozen-lockfile
- name: Check linting
run: cd src/frontend/ && yarn lint
run: yarn lint
test-e2e:
runs-on: ubuntu-latest
needs: [build-mails, build-front]
timeout-minutes: 10
steps:
- name: Checkout repository
@@ -154,28 +115,7 @@ jobs:
- name: Set services env variables
run: |
make create-env-files
cat env.d/development/common.e2e.dist >> env.d/development/common
- name: Download mails' templates
uses: actions/download-artifact@v4
with:
name: mails-templates
path: src/backend/core/templates/mail
- name: Restore the frontend cache
uses: actions/cache@v4
id: front-node_modules
with:
path: 'src/frontend/**/node_modules'
key: front-node_modules-${{ hashFiles('src/frontend/**/yarn.lock') }}
- name: Restore the build cache
uses: actions/cache@v4
id: cache-build
with:
path: src/frontend/apps/desk/out/
key: build-front-${{ github.run_id }}
- name: Build and Start Docker Servers
env:
DOCKER_BUILDKIT: 1
@@ -184,21 +124,30 @@ jobs:
docker-compose build --pull --build-arg BUILDKIT_INLINE_CACHE=1
make run
- name: Apply DRF migrations
- name: Apply DRH migrations
run: |
make migrate
- name: Add dummy data
run: |
make demo FLUSH_ARGS='--no-input'
- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: '18.x'
cache: 'yarn'
cache-dependency-path: src/frontend/yarn.lock
- name: Install dependencies
run: cd src/frontend/ && yarn install --frozen-lockfile
- name: Install Playwright Browsers
run: cd src/frontend/apps/e2e && yarn install
- name: Build CI App
run: cd src/frontend/ && yarn ci:build
- name: Run e2e tests
run: cd src/frontend/ && yarn e2e:test
- uses: actions/upload-artifact@v4
- uses: actions/upload-artifact@v3
if: always()
with:
name: playwright-report
@@ -212,7 +161,7 @@ jobs:
working-directory: src/mail
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
- name: Install Node.js
uses: actions/setup-node@v4
with:
@@ -223,11 +172,6 @@ jobs:
run: yarn install --frozen-lockfile
- name: Build mails
run: yarn build
- name: Persist mails' templates
uses: actions/upload-artifact@v4
with:
name: mails-templates
path: src/backend/core/templates/mail
lint-back:
runs-on: ubuntu-latest
@@ -236,9 +180,9 @@ jobs:
working-directory: src/backend
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
- name: Install Python
uses: actions/setup-python@v5
uses: actions/setup-python@v3
with:
python-version: '3.10'
- name: Install development dependencies
@@ -252,7 +196,6 @@ jobs:
test-back:
runs-on: ubuntu-latest
needs: build-mails
defaults:
run:
working-directory: src/backend
@@ -273,7 +216,7 @@ jobs:
DJANGO_CONFIGURATION: Test
DJANGO_SETTINGS_MODULE: people.settings
DJANGO_SECRET_KEY: ThisIsAnExampleKeyForTestPurposeOnly
OIDC_OP_JWKS_ENDPOINT: /endpoint-for-test-purpose-only
DJANGO_JWT_PRIVATE_SIGNING_KEY: ThisIsAnExampleKeyForDevPurposeOnly
DB_HOST: localhost
DB_NAME: people
DB_USER: dinum
@@ -282,18 +225,13 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v4
uses: actions/checkout@v2
- name: Create writable /data
run: |
sudo mkdir -p /data/media && \
sudo mkdir -p /data/static
- name: Download mails' templates
uses: actions/download-artifact@v4
with:
name: mails-templates
path: src/backend/core/templates/mail
- name: Install Python
uses: actions/setup-python@v5
uses: actions/setup-python@v3
with:
python-version: '3.10'
- name: Install development dependencies
@@ -310,27 +248,16 @@ jobs:
i18n-crowdin:
runs-on: ubuntu-latest
steps:
-
uses: actions/create-github-app-token@v1
id: app-token
with:
app-id: ${{ secrets.APP_ID }}
private-key: ${{ secrets.PRIVATE_KEY }}
owner: ${{ github.repository_owner }}
repositories: "people,secrets"
-
name: Checkout repository
- name: Checkout repository
uses: actions/checkout@v2
with:
submodules: recursive
token: ${{ steps.app-token.outputs.token }}
- name: Install gettext (required to make messages)
run: |
sudo apt-get update
sudo apt-get install -y gettext
- name: Install Python
uses: actions/setup-python@v5
uses: actions/setup-python@v3
with:
python-version: '3.10'
@@ -341,11 +268,10 @@ jobs:
- name: Generate the translation base file
run: ~/.local/bin/django-admin makemessages --keep-pot --all
-
name: Load sops secrets
- name: Load sops secrets
uses: rouja/actions-sops@main
with:
secret-file: secrets/numerique-gouv/people/secrets.enc.env
secret-file: .github/workflows/secrets.enc.env
age-key: ${{ secrets.SOPS_PRIVATE }}
- name: Setup Node.js
+20
View File
@@ -0,0 +1,20 @@
SOPS_PRIVATE=ENC[AES256_GCM,data:uMsnbpi1FFxO430iptp2axlH/souCPCJ/afCqh/kIDWs8xWHu0xJ2o6PlNOgb4l1gHF8sy4eSHFOW5HPKVbZ9gafRIL0JYHJr3A=,iv:IgaHDad7IuW2wFoxGALLDCs+UiSd3rEYwGNSX38wUfI=,tag:zs4wn4hDm5fghfI/7iH6CQ==,type:str]
CROWDIN_API_TOKEN=ENC[AES256_GCM,data:tZRGFs792GjKYGit+oNubCwPbjWarhlgcyQ7oStO8K3nao0lEzLrm3+ARaOEXkzEkKSal2N1c+vlYIjRyzV1X7WZ7nZQWBiihEJgAfF3NGM=,iv:hohak/1niu5kV/W4XA0KEE3UB0BuNCDxbRxb0O+Aocs=,tag:Ssi0HyEhCKb9+WFyp3/yBQ==,type:str]
CROWDIN_BASE_PATH=ENC[AES256_GCM,data:m44KrW5xJDE=,iv:bSyKrKQ0Z1yzrNaejAUyD+n1Z9z+ci92GoyS5KiiFKI=,tag:dDtpbn+6LkPwsg+qgMNWvA==,type:str]
CROWDIN_PROJECT_ID=ENC[AES256_GCM,data:Gonh6ps7,iv:yUqPty+R060m6CYrDf7na2f6h0aRpIhRCXZoJW4ThJg=,tag:ewupoenajsc9o0Aj0r/niw==,type:str]
DOCKER_HUB_PASSWORD=ENC[AES256_GCM,data:CBjUvS/UQcqdEv8=,iv:zJKGyyRnq+zxhKkYS5h3zw/J1320zrAqpiObcwiHWsA=,tag:OoQNe+4K63KXTWNsvjRZNQ==,type:str]
DOCKER_HUB_USER=ENC[AES256_GCM,data:ivs7oeBBdA==,iv:cKAQJ071XhcRuFyeHhZkscvvz7sHrKIPJg5pt76hCXg=,tag:2VRmp1ULWvVH8NHcBifBGA==,type:str]
sops_age__list_0__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBIbkIyRi9jTGEzL2RZcGFS\nVkhuN2xMN1FBcm04eGJQUnpvQVRpdmFmaFF3CjdoNVhSM0lrb0lTSXAyUytSKzhG\nNERYaC80OThmUUxCSmt2U2Q0Mno3b0kKLS0tIFRjQzYyOEcwTnRScGhzaGxwaGFL\namxEUXNaeVhWUjYwbzR2Z1Uzb3JsTlEK6tyhbRmcUP4Aql49DPkrYb5tbwvK2EdA\ntvyPQo4pPit+pzgqsVgW+O8Wo4/rYLlITfuVRrOfHEaH3wmf6hziSw==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_0__map_recipient=age15fyxdwmg5mvldtqqus87xspuws2u0cpvwheehrtvkexj4tnsqqysw6re2x
sops_age__list_1__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBIL2NwVjVtVmd1ZE15WmFC\nVGl5cnU1OHVUcXAxWmZCQXBJbXBlcXFUN0RzClpOVXBTek9wbVB6M2s5TWlFbUo3\nelZrT3dsK3p6cEJGTGJoSTRvaEh4NW8KLS0tIGlsL29oSEthU28xV0RERFIxTnZR\nSSs3YjdUT0NyTEtlbjlmZXVOaTd3SzgK5jFJGREfJ/HVytWsCKWFsqM5JoaFSnhv\n538KzzldzcbtWfnY+bQ6A2EBjETwOzCTuQB8axAMj0URXPI+qelKIQ==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_1__map_recipient=age16hnlml8yv4ynwy0seer57g8qww075crd0g7nsundz3pj4wk7m3vqftszg7
sops_age__list_2__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBzOWtwSytMNGRmSHNQN0Zz\nV1RTWktXT1c5VjYvNTg2V0x1U3JSTFA1dmtvCnc4RHRodjVxSzcrZXJycCtJMGNy\nZ3ZwbTFDTUc4TTMxbm8rNXQ1ME90Q1UKLS0tIERPUmg1dSt5OWQwaXV0Y2FiYWdt\nVzlhcFNvUS9Nd3A5ZTYwT2lLUU5WVU0KGHSm1BQ3voKs98WiXNLO6hlqoiQmi1/F\n2RCBkE4/gOVyAmJAjOFizaF0Dhd7Ba4KS5l5QylFHloL8XtyixEhog==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_2__map_recipient=age1g5keveae6zhn059e7cxkjqdz4v3u47ypejv9ujld65nwh6d5pd9qfm0ecv
sops_age__list_3__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBmYlB6cEZKK3NLbm5Ob0hi\nS3M0OFoyTHhVYVA1WmcvSHc2MUErQ2JyaUh3Cnp1NlYvU1BPUnVON29UZWUvWmEz\nOW85N013VUJ3Q1ZZUmMvTWdYclBZTUkKLS0tIG1KODdkazhTUHRPMXZXQnJFNXJY\nOFlSNjZCbkpxcm9tN1dFTkZ4K1lETFUKhiPwKEG71OlTcK/Ul1GKGayLy025vAmo\nNgQUhbqXf7KmqDAmrQNzXTsLsOpRi33l66jFSGkTsFtiXNlmjFljKg==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_3__map_recipient=age12g6f5fse25tgrwweleh4jls3qs52hey2edh759smulwmk5lnzadslu2cp3
sops_age__list_4__map_enc=-----BEGIN AGE ENCRYPTED FILE-----\nYWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBJUzI2eDB0MU41YWoxcjAv\nOU8yMlowb3IybCtzcFBBTEVlUzdBczZkQlFNCk5tbnhLR2kweGNFZmx1OFgwaWZU\nTzRKZTBZamJjRFNYOUJlYmxTUVg4S2MKLS0tIDNGdlVVc0hLK1BPTUJVc0tPb1lK\ndVVWVi9GRkxwYXR1cXMrdnlsejJBeGsKKmVWvIMrBYH+UrDMkZPxN8KWnCgA6WK2\nbexMYr2AIF2QMbhck7XW2NuvAwvwjbJMfcd+cp9boe+EcC4YjdJZlw==\n-----END AGE ENCRYPTED FILE-----\n
sops_age__list_4__map_recipient=age1tl80n23wq6zxegupwn70ew0yp225ua5v4dk800x7g2w6pvlxz46qk592pa
sops_lastmodified=2024-02-05T17:45:08Z
sops_mac=ENC[AES256_GCM,data:WMsYYvGId0UlRhiw5C4TTpP/7oIIgWteGRtX7Nlo3qQwBc1LSUyN+7kblkXWD+nNG51B0xfl9E5QdXO3Ao2WskQZZEJqqkmt6wSw/ScSPcH56FoIqBhoGsPKyJOZkK0zojnCJQLniTDlISG0UiiJ4KHh8FJDWYbzAy2zFEm3Et8=,iv:zB2hyxnKiA3flsMQ7XFOT/fR9hamd6YWo3BNwce4oWM=,tag:uZ2Ia8sR5R8fmURtJ+PojA==,type:str]
sops_unencrypted_suffix=_unencrypted
sops_version=3.8.1
-1
View File
@@ -79,4 +79,3 @@ db.sqlite3
.vscode/
*.iml
.devcontainer/
.tool-versions
-3
View File
@@ -1,3 +0,0 @@
[submodule "secrets"]
path = secrets
url = ../secrets
+13
View File
@@ -0,0 +1,13 @@
creation_rules:
# Here we have
# - Jacques key-id: age15fyxdwmg5mvldtqqus87xspuws2u0cpvwheehrtvkexj4tnsqqysw6re2x
# - github-repo key-id: age16hnlml8yv4ynwy0seer57g8qww075crd0g7nsundz3pj4wk7m3vqftszg7
# - Anthony Le-Courric key-id: age1g5keveae6zhn059e7cxkjqdz4v3u47ypejv9ujld65nwh6d5pd9qfm0ecv
# - Antoine Lebaud key-id: age12g6f5fse25tgrwweleh4jls3qs52hey2edh759smulwmk5lnzadslu2cp3
# - Marie Pupo Jeammet key-id: age1tl80n23wq6zxegupwn70ew0yp225ua5v4dk800x7g2w6pvlxz46qk592pa
- age:
age15fyxdwmg5mvldtqqus87xspuws2u0cpvwheehrtvkexj4tnsqqysw6re2x,
age16hnlml8yv4ynwy0seer57g8qww075crd0g7nsundz3pj4wk7m3vqftszg7,
age1g5keveae6zhn059e7cxkjqdz4v3u47ypejv9ujld65nwh6d5pd9qfm0ecv,
age12g6f5fse25tgrwweleh4jls3qs52hey2edh759smulwmk5lnzadslu2cp3,
age1tl80n23wq6zxegupwn70ew0yp225ua5v4dk800x7g2w6pvlxz46qk592pa
+16 -31
View File
@@ -11,46 +11,31 @@ RUN apt-get update && \
apt-get -y upgrade && \
rm -rf /var/lib/apt/lists/*
### ---- Front-end dependencies image ----
FROM node:20 as frontend-deps
WORKDIR /deps
COPY ./src/frontend/package.json ./package.json
COPY ./src/frontend/yarn.lock ./yarn.lock
COPY ./src/frontend/apps/desk/package.json ./apps/desk/package.json
COPY ./src/frontend/packages/i18n/package.json ./packages/i18n/package.json
COPY ./src/frontend/packages/eslint-config-people/package.json ./packages/eslint-config-people/package.json
RUN yarn --frozen-lockfile
### ---- Front-end builder dev image ----
FROM node:20 as frontend-builder-dev
WORKDIR /builder
COPY --from=frontend-deps /deps/node_modules ./node_modules
COPY ./src/frontend .
WORKDIR ./apps/desk
### ---- Front-end builder image ----
FROM frontend-builder-dev as frontend-builder
FROM node:20 as front-builder
# Copy frontend app sources
COPY ./src/frontend /builder
WORKDIR /builder/apps/desk
RUN yarn install --frozen-lockfile && \
yarn build
RUN yarn build
# ---- Front-end image ----
FROM nginxinc/nginx-unprivileged:1.25 as frontend-production
FROM nginxinc/nginx-unprivileged:1.25 as frontend
# Un-privileged user running the application
ARG DOCKER_USER
USER ${DOCKER_USER}
COPY --from=frontend-builder \
COPY --from=front-builder \
/builder/apps/desk/out \
/usr/share/nginx/html
COPY ./src/frontend/apps/desk/conf/default.conf /etc/nginx/conf.d
COPY ./docker/files/etc/nginx/conf.d /etc/nginx/conf.d:ro
# Copy entrypoint
COPY ./docker/files/usr/local/bin/entrypoint /usr/local/bin/entrypoint
@@ -149,7 +134,7 @@ WORKDIR /app
ENTRYPOINT [ "/usr/local/bin/entrypoint" ]
# ---- Development image ----
FROM core as backend-development
FROM core as development
# Switch back to the root user to install development dependencies
USER root:root
@@ -174,10 +159,10 @@ ENV DB_HOST=postgresql \
DB_PORT=5432
# Run django development server
CMD ["python", "manage.py", "runserver", "0.0.0.0:8000"]
CMD python manage.py runserver 0.0.0.0:8000
# ---- Production image ----
FROM core as backend-production
FROM core as production
ARG PEOPLE_STATIC_ROOT=/data/static
@@ -196,4 +181,4 @@ COPY --from=link-collector ${PEOPLE_STATIC_ROOT} ${PEOPLE_STATIC_ROOT}
COPY --from=mail-builder /mail/backend/core/templates/mail /app/core/templates/mail
# The default command runs gunicorn WSGI server in people's main module
CMD ["gunicorn", "-c", "/usr/local/etc/gunicorn/people.py", "people.wsgi:application"]
CMD gunicorn -c /usr/local/etc/gunicorn/people.py people.wsgi:application
+8 -22
View File
@@ -92,8 +92,10 @@ bootstrap: \
.PHONY: bootstrap
# -- Docker/compose
build: ## build the app-dev container
build: ## build the dev and demo containers
@$(COMPOSE) build app-dev
@$(COMPOSE) build nginx
@$(COMPOSE) build app
.PHONY: build
down: ## stop and remove containers, networks, images, and volumes
@@ -166,30 +168,24 @@ test-back-parallel: ## run all back-end tests in parallel
bin/pytest -n auto $${args:-${1}}
.PHONY: test-back-parallel
makemigrations: ## run django makemigrations for the people project.
@echo "$(BOLD)Running makemigrations$(RESET)"
@$(COMPOSE) up -d postgresql
@$(WAIT_DB)
@$(MANAGE) makemigrations $(ARGS)
@$(MANAGE) makemigrations
.PHONY: makemigrations
migrate: ## run django migrations for the people project.
@echo "$(BOLD)Running migrations$(RESET)"
@$(COMPOSE) up -d postgresql
@$(WAIT_DB)
@$(MANAGE) migrate $(ARGS)
@$(MANAGE) migrate
.PHONY: migrate
showmigrations: ## run django showmigrations for the people project.
@echo "$(BOLD)Running showmigrations$(RESET)"
@$(COMPOSE) up -d postgresql
@$(WAIT_DB)
@$(MANAGE) showmigrations $(ARGS)
.PHONY: showmigrations
superuser: ## Create an admin superuser with password "admin"
@echo "$(BOLD)Creating a Django superuser$(RESET)"
@$(MANAGE) createsuperuser --admin_email admin@example.com --password admin
@$(MANAGE) createsuperuser --email admin@example.com --password admin
.PHONY: superuser
back-i18n-compile: ## compile the gettext files
@@ -210,10 +206,9 @@ dbshell: ## connect to database shell
docker compose exec app-dev python manage.py dbshell
.PHONY: dbshell
resetdb: FLUSH_ARGS ?=
resetdb: ## flush database and create a superuser "admin"
@echo "$(BOLD)Flush database$(RESET)"
@$(MANAGE) flush $(FLUSH_ARGS)
@$(MANAGE) flush
@${MAKE} superuser
.PHONY: resetdb
@@ -329,12 +324,3 @@ frontend-i18n-generate: \
frontend-i18n-compile: ## Format the crowin json files used deploy to the apps
cd $(PATH_FRONT) && yarn i18n:deploy
.PHONY: frontend-i18n-compile
# -- K8S
start-kind: ## Create the kubernetes cluster
./bin/start-kind.sh
.PHONY: start-kind
tilt-up: ## start tilt - k8s local development
tilt up -f ./bin/Tiltfile
.PHONY: tilt-up
-18
View File
@@ -69,24 +69,6 @@ You first need to create a superuser account:
$ make superuser
```
You can then login with credentials `admin@example` / `admin`.
### Run frontend
Run the front with:
```bash
$ make run-front-desk
```
Then access at
[http://localhost:3000](http://localhost:3000)
user: people
password: people
## Contributing
This project is intended to be community-driven, so please, do not hesitate to
-58
View File
@@ -1,58 +0,0 @@
load('ext://uibutton', 'cmd_button', 'bool_input', 'location')
load('ext://namespace', 'namespace_create', 'namespace_inject')
namespace_create('desk')
docker_build(
'localhost:5001/people-backend:latest',
context='..',
dockerfile='../Dockerfile',
only=['./src/backend', './src/mail', './docker'],
target = 'backend-production',
live_update=[
sync('../src/backend', '/app'),
run(
'pip install -r /app/requirements.txt',
trigger=['./api/requirements.txt']
)
]
)
docker_build(
'localhost:5001/people-frontend:latest',
context='..',
dockerfile='../Dockerfile',
build_args={'ENV': 'dev'},
only=['./src/frontend', './src/mail', './docker'],
target = 'frontend-builder-dev',
live_update=[
sync('../src/frontend', '/builder'),
]
)
k8s_yaml(local('cd ../src/helm && helmfile -n desk -e dev template .'))
migration = '''
set -eu
# get k8s pod name from tilt resource name
POD_NAME="$(tilt get kubernetesdiscovery desk-backend -ojsonpath='{.status.pods[0].name}')"
kubectl -n desk exec "$POD_NAME" -- python manage.py makemigrations
'''
cmd_button('Make migration',
argv=['sh', '-c', migration],
resource='desk-backend',
icon_name='developer_board',
text='Run makemigration',
)
pod_migrate = '''
set -eu
# get k8s pod name from tilt resource name
POD_NAME="$(tilt get kubernetesdiscovery desk-backend -ojsonpath='{.status.pods[0].name}')"
kubectl -n desk exec "$POD_NAME" -- python manage.py migrate --no-input
'''
cmd_button('Migrate db',
argv=['sh', '-c', pod_migrate],
resource='desk-backend',
icon_name='developer_board',
text='Run database migration',
)
-102
View File
@@ -1,102 +0,0 @@
#!/bin/sh
set -o errexit
CURRENT_DIR=$(pwd)
# 0. Create ca
echo "0. Create ca"
mkcert -install
cd /tmp
mkcert "127.0.0.1.nip.io" "*.127.0.0.1.nip.io"
cd $CURRENT_DIR
# 1. Create registry container unless it already exists
echo "1. Create registry container unless it already exists"
reg_name='kind-registry'
reg_port='5001'
if [ "$(docker inspect -f '{{.State.Running}}' "${reg_name}" 2>/dev/null || true)" != 'true' ]; then
docker run \
-d --restart=always -p "127.0.0.1:${reg_port}:5000" --network bridge --name "${reg_name}" \
registry:2
fi
# 2. Create kind cluster with containerd registry config dir enabled
echo "2. Create kind cluster with containerd registry config dir enabled"
# TODO: kind will eventually enable this by default and this patch will
# be unnecessary.
#
# See:
# https://github.com/kubernetes-sigs/kind/issues/2875
# https://github.com/containerd/containerd/blob/main/docs/cri/config.md#registry-configuration
# See: https://github.com/containerd/containerd/blob/main/docs/hosts.md
cat <<EOF | kind create cluster --config=-
kind: Cluster
apiVersion: kind.x-k8s.io/v1alpha4
containerdConfigPatches:
- |-
[plugins."io.containerd.grpc.v1.cri".registry]
config_path = "/etc/containerd/certs.d"
nodes:
- role: control-plane
image: kindest/node:v1.27.3
kubeadmConfigPatches:
- |
kind: InitConfiguration
nodeRegistration:
kubeletExtraArgs:
node-labels: "ingress-ready=true"
extraPortMappings:
- containerPort: 80
hostPort: 80
protocol: TCP
- containerPort: 443
hostPort: 443
protocol: TCP
- role: worker
image: kindest/node:v1.27.3
- role: worker
image: kindest/node:v1.27.3
EOF
# 3. Add the registry config to the nodes
echo "3. Add the registry config to the nodes"
#
# This is necessary because localhost resolves to loopback addresses that are
# network-namespace local.
# In other words: localhost in the container is not localhost on the host.
#
# We want a consistent name that works from both ends, so we tell containerd to
# alias localhost:${reg_port} to the registry container when pulling images
REGISTRY_DIR="/etc/containerd/certs.d/localhost:${reg_port}"
for node in $(kind get nodes); do
docker exec "${node}" mkdir -p "${REGISTRY_DIR}"
cat <<EOF | docker exec -i "${node}" cp /dev/stdin "${REGISTRY_DIR}/hosts.toml"
[host."http://${reg_name}:5000"]
EOF
done
# 4. Connect the registry to the cluster network if not already connected
echo "4. Connect the registry to the cluster network if not already connected"
# This allows kind to bootstrap the network but ensures they're on the same network
if [ "$(docker inspect -f='{{json .NetworkSettings.Networks.kind}}' "${reg_name}")" = 'null' ]; then
docker network connect "kind" "${reg_name}"
fi
# 5. Document the local registry
echo "5. Document the local registry"
# https://github.com/kubernetes/enhancements/tree/master/keps/sig-cluster-lifecycle/generic/1755-communicating-a-local-registry
cat <<EOF | kubectl apply -f -
apiVersion: v1
kind: ConfigMap
metadata:
name: local-registry-hosting
namespace: kube-public
data:
localRegistryHosting.v1: |
host: "localhost:${reg_port}"
help: "https://kind.sigs.k8s.io/docs/user/local-registry/"
EOF
kubectl apply -f https://raw.githubusercontent.com/kubernetes/ingress-nginx/main/deploy/static/provider/kind/deploy.yaml
kubectl -n ingress-nginx create secret tls mkcert --key /tmp/127.0.0.1.nip.io+1-key.pem --cert /tmp/127.0.0.1.nip.io+1.pem
kubectl -n ingress-nginx patch deployments.apps ingress-nginx-controller --type 'json' -p '[{"op": "add", "path": "/spec/template/spec/containers/0/args/-", "value":"--default-ssl-certificate=ingress-nginx/mkcert"}]'
+16 -8
View File
@@ -19,11 +19,11 @@ services:
app-dev:
build:
context: .
target: backend-development
target: development
args:
DOCKER_USER: ${DOCKER_USER:-1000}
user: ${DOCKER_USER:-1000}
image: people:backend-development
image: people:development
environment:
- PYLINTHOME=/app/.pylint.d
- DJANGO_CONFIGURATION=Development
@@ -43,7 +43,7 @@ services:
celery-dev:
user: ${DOCKER_USER:-1000}
image: people:backend-development
image: people:development
command: ["celery", "-A", "people.celery_app", "worker", "-l", "DEBUG"]
environment:
- DJANGO_CONFIGURATION=Development
@@ -60,16 +60,19 @@ services:
app:
build:
context: .
target: backend-production
target: production
args:
DOCKER_USER: ${DOCKER_USER:-1000}
user: ${DOCKER_USER:-1000}
image: people:backend-production
image: people:production
environment:
- DJANGO_CONFIGURATION=Demo
- DJANGO_CORS_ALLOWED_ORIGINS=http://localhost:8088
env_file:
- env.d/development/common
- env.d/development/postgresql
ports:
- "8082:8000"
volumes:
- ./data/media:/data/media
depends_on:
@@ -78,7 +81,7 @@ services:
celery:
user: ${DOCKER_USER:-1000}
image: people:backend-production
image: people:production
command: ["celery", "-A", "people.celery_app", "worker", "-l", "INFO"]
environment:
- DJANGO_CONFIGURATION=Demo
@@ -89,8 +92,12 @@ services:
- app
nginx:
image: nginx:1.25
build:
context: .
target: frontend
image: people:frontend
ports:
- "8088:8088"
- "8083:8083"
volumes:
- ./docker/files/etc/nginx/conf.d:/etc/nginx/conf.d:ro
@@ -136,6 +143,7 @@ services:
kc_postgresql:
image: postgres:14.3
platform: linux/amd64
ports:
- "5433:5432"
env_file:
@@ -163,7 +171,7 @@ services:
KC_DB_PASSWORD: pass
KC_DB_USERNAME: people
KC_DB_SCHEMA: public
PROXY_ADDRESS_FORWARDING: 'true'
PROXY_ADDRESS_FORWARDING: true
ports:
- "8080:8080"
depends_on:
+6 -40
View File
@@ -46,9 +46,6 @@
"users": [
{
"username": "people",
"email": "people@people.world",
"firstName": "John",
"lastName": "Doe",
"enabled": true,
"credentials": [
{
@@ -59,43 +56,12 @@
"realmRoles": ["user"]
},
{
"username": "user-e2e-chromium",
"email": "user@chromium.e2e",
"firstName": "E2E",
"lastName": "Chromium",
"username": "user-e2e",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e-chromium"
}
],
"realmRoles": ["user"]
},
{
"username": "user-e2e-webkit",
"email": "user@webkit.e2e",
"firstName": "E2E",
"lastName": "Webkit",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e-webkit"
}
],
"realmRoles": ["user"]
},
{
"username": "user-e2e-firefox",
"email": "user@firefox.e2e",
"firstName": "E2E",
"lastName": "Firefox",
"enabled": true,
"credentials": [
{
"type": "password",
"value": "password-e2e-firefox"
"value": "password-e2e"
}
],
"realmRoles": ["user"]
@@ -1110,7 +1076,7 @@
},
{
"id": "79712bcf-b7f7-4ca3-b97c-418f48fded9b",
"name": "first name",
"name": "given name",
"protocol": "openid-connect",
"protocolMapper": "oidc-usermodel-property-mapper",
"consentRequired": false,
@@ -1119,7 +1085,7 @@
"user.attribute": "firstName",
"id.token.claim": "true",
"access.token.claim": "true",
"claim.name": "first_name",
"claim.name": "given_name",
"jsonType.label": "String"
}
},
@@ -1140,7 +1106,7 @@
},
{
"id": "7f741e96-41fe-4021-bbfd-506e7eb94e69",
"name": "last name",
"name": "family name",
"protocol": "openid-connect",
"protocolMapper": "oidc-usermodel-property-mapper",
"consentRequired": false,
@@ -1149,7 +1115,7 @@
"user.attribute": "lastName",
"id.token.claim": "true",
"access.token.claim": "true",
"claim.name": "last_name",
"claim.name": "family_name",
"jsonType.label": "String"
}
},
@@ -1,3 +1,19 @@
server {
listen 8088;
server_name localhost;
root /usr/share/nginx/html;
location / {
try_files $uri index.html $uri/ =404;
}
error_page 404 /404.html;
location = /404.html {
internal;
}
}
server {
listen 8083;
@@ -11,3 +27,4 @@ server {
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}
-3
View File
@@ -1,3 +0,0 @@
# For the CI job test-e2e
SUSTAINED_THROTTLE_RATES="200/hour"
BURST_THROTTLE_RATES="200/minute"
+1 -1
View File
@@ -8,4 +8,4 @@ DB_HOST=kc_postgresql
DB_NAME=keycloak
DB_USER=people
DB_PASSWORD=pass
DB_PORT=5433
DB_PORT=5433
+1 -1
View File
@@ -8,4 +8,4 @@ DB_HOST=postgresql
DB_NAME=people
DB_USER=dinum
DB_PASSWORD=pass
DB_PORT=5432
DB_PORT=5432
+1 -2
View File
@@ -1,7 +1,6 @@
{
"extends": ["github>numerique-gouv/renovate-configuration"],
"dependencyDashboard": true,
"labels": ["dependencies", "noChangeLog"],
"packageRules": [
{
"enabled": false,
@@ -13,7 +12,7 @@
"enabled": false,
"groupName": "ignored js dependencies",
"matchManagers": ["npm"],
"matchPackageNames": ["node", "node-fetch", "i18next-parser", "eslint"]
"matchPackageNames": ["node", "node-fetch", "i18next-parser"]
}
]
}
-4
View File
@@ -1,4 +0,0 @@
#!/bin/bash
git submodule update --init --recursive
git submodule foreach 'git fetch origin; git checkout $(git rev-parse --abbrev-ref HEAD); git reset --hard origin/$(git rev-parse --abbrev-ref HEAD); git submodule update --recursive; git clean -dfx'
Submodule secrets deleted from d7cfe7bcdc
+4 -65
View File
@@ -1,5 +1,4 @@
"""Admin classes and registrations for People's core app."""
from django import forms
from django.contrib import admin
from django.contrib.auth import admin as auth_admin
@@ -53,15 +52,6 @@ class TeamAccessInline(admin.TabularInline):
readonly_fields = ("created_at", "updated_at")
class TeamWebhookInline(admin.TabularInline):
"""Inline admin class for team webhooks."""
extra = 0
autocomplete_fields = ["team"]
model = models.TeamWebhook
readonly_fields = ("created_at", "updated_at")
@admin.register(models.User)
class UserAdmin(auth_admin.UserAdmin):
"""Admin class for the User model"""
@@ -76,7 +66,7 @@ class UserAdmin(auth_admin.UserAdmin):
)
},
),
(_("Personal info"), {"fields": ("admin_email", "language", "timezone")}),
(_("Personal info"), {"fields": ("email", "language", "timezone")}),
(
_("Permissions"),
{
@@ -92,18 +82,9 @@ class UserAdmin(auth_admin.UserAdmin):
),
(_("Important dates"), {"fields": ("created_at", "updated_at")}),
)
add_fieldsets = (
(
None,
{
"classes": ("wide",),
"fields": ("admin_email", "password1", "password2"),
},
),
)
inlines = (IdentityInline, TeamAccessInline)
list_display = (
"admin_email",
"email",
"created_at",
"updated_at",
"is_active",
@@ -114,14 +95,14 @@ class UserAdmin(auth_admin.UserAdmin):
list_filter = ("is_staff", "is_superuser", "is_device", "is_active")
ordering = ("is_active", "-is_superuser", "-is_staff", "-is_device", "-updated_at")
readonly_fields = ("id", "created_at", "updated_at")
search_fields = ("id", "admin_email", "identities__sub", "identities__email")
search_fields = ("id", "email", "identities__sub", "identities__email")
@admin.register(models.Team)
class TeamAdmin(admin.ModelAdmin):
"""Team admin interface declaration."""
inlines = (TeamAccessInline, TeamWebhookInline)
inlines = (TeamAccessInline,)
list_display = (
"name",
"slug",
@@ -129,45 +110,3 @@ class TeamAdmin(admin.ModelAdmin):
"updated_at",
)
search_fields = ("name",)
@admin.register(models.Invitation)
class InvitationAdmin(admin.ModelAdmin):
"""Admin interface to handle invitations."""
fields = (
"email",
"team",
"role",
"created_at",
"issuer",
)
readonly_fields = (
"created_at",
"is_expired",
"issuer",
)
list_display = (
"email",
"team",
"created_at",
"is_expired",
)
def get_readonly_fields(self, request, obj=None):
if obj:
# all fields read only = disable update
return self.fields
return self.readonly_fields
def change_view(self, request, object_id, form_url="", extra_context=None):
"""Custom edit form. Remove 'save' buttons."""
extra_context = extra_context or {}
extra_context["show_save_and_continue"] = False
extra_context["show_save"] = False
extra_context["show_save_and_add_another"] = False
return super().change_view(request, object_id, extra_context=extra_context)
def save_model(self, request, obj, form, change):
obj.issuer = request.user
obj.save()
-3
View File
@@ -1,5 +1,4 @@
"""People core API endpoints"""
from django.conf import settings
from django.core.exceptions import ValidationError
@@ -23,8 +22,6 @@ def exception_handler(exc, context):
detail = exc.message
elif hasattr(exc, "messages"):
detail = exc.messages
else:
detail = ""
exc = drf_exceptions.ValidationError(detail=detail)
-1
View File
@@ -1,5 +1,4 @@
"""Permission handlers for the People core app."""
from django.core import exceptions
from rest_framework import permissions
+11 -100
View File
@@ -1,5 +1,4 @@
"""Client serializers for the People core app."""
from rest_framework import exceptions, serializers
from timezone_field.rest_framework import TimeZoneSerializerField
@@ -27,46 +26,28 @@ class ContactSerializer(serializers.ModelSerializer):
return super().update(instance, validated_data)
class DynamicFieldsModelSerializer(serializers.ModelSerializer):
"""
A ModelSerializer that takes an additional `fields` argument that
controls which fields should be displayed.
"""
def __init__(self, *args, **kwargs):
# Don't pass the 'fields' arg up to the superclass
fields = kwargs.pop("fields", None)
# Instantiate the superclass normally
super().__init__(*args, **kwargs)
if fields is not None:
# Drop any fields that are not specified in the `fields` argument.
allowed = set(fields)
existing = set(self.fields)
for field_name in existing - allowed:
self.fields.pop(field_name)
class UserSerializer(DynamicFieldsModelSerializer):
class UserSerializer(serializers.ModelSerializer):
"""Serialize users."""
data = serializers.SerializerMethodField(read_only=True)
timezone = TimeZoneSerializerField(use_pytz=False, required=True)
email = serializers.ReadOnlyField()
name = serializers.ReadOnlyField()
class Meta:
model = models.User
fields = [
"id",
"email",
"data",
"language",
"name",
"timezone",
"is_device",
"is_staff",
]
read_only_fields = ["id", "name", "email", "is_device", "is_staff"]
read_only_fields = ["id", "email", "data", "is_device", "is_staff"]
def get_data(self, user) -> dict:
"""Return contact data for the user."""
return user.profile_contact.data if user.profile_contact else {}
class TeamAccessSerializer(serializers.ModelSerializer):
@@ -145,52 +126,17 @@ class TeamAccessSerializer(serializers.ModelSerializer):
return attrs
class TeamAccessReadOnlySerializer(TeamAccessSerializer):
"""Serialize team accesses for list and retrieve actions."""
user = UserSerializer(read_only=True, fields=["id", "name", "email"])
class Meta:
model = models.TeamAccess
fields = [
"id",
"user",
"role",
"abilities",
]
read_only_fields = [
"id",
"user",
"role",
"abilities",
]
class TeamSerializer(serializers.ModelSerializer):
"""Serialize teams."""
abilities = serializers.SerializerMethodField(read_only=True)
accesses = TeamAccessSerializer(many=True, read_only=True)
slug = serializers.SerializerMethodField()
class Meta:
model = models.Team
fields = [
"id",
"name",
"accesses",
"abilities",
"slug",
"created_at",
"updated_at",
]
read_only_fields = [
"id",
"accesses",
"abilities",
"slug",
"created_at",
"updated_at",
]
fields = ["id", "name", "accesses", "abilities", "slug"]
read_only_fields = ["id", "accesses", "abilities", "slug"]
def get_abilities(self, team) -> dict:
"""Return abilities of the logged-in user on the instance."""
@@ -202,38 +148,3 @@ class TeamSerializer(serializers.ModelSerializer):
def get_slug(self, instance):
"""Return slug from the team's name."""
return instance.get_slug()
class InvitationSerializer(serializers.ModelSerializer):
"""Serialize invitations."""
class Meta:
model = models.Invitation
fields = ["id", "created_at", "email", "team", "role", "issuer", "is_expired"]
read_only_fields = ["id", "created_at", "team", "issuer", "is_expired"]
def validate(self, attrs):
"""Validate and restrict invitation to new user based on email."""
request = self.context.get("request")
user = getattr(request, "user", None)
try:
team_id = self.context["team_id"]
except KeyError as exc:
raise exceptions.ValidationError(
"You must set a team ID in kwargs to create a new team invitation."
) from exc
if not models.TeamAccess.objects.filter(
team=team_id,
user=user,
role__in=[models.RoleChoices.OWNER, models.RoleChoices.ADMIN],
).exists():
raise exceptions.PermissionDenied(
"You are not allowed to manage invitation for this team."
)
attrs["team_id"] = team_id
attrs["issuer"] = user
return attrs
+14
View File
@@ -0,0 +1,14 @@
"""
Utils that can be useful throughout the People core app
"""
from rest_framework_simplejwt.tokens import RefreshToken
def get_tokens_for_user(user):
"""Get JWT tokens for user authentication."""
refresh = RefreshToken.for_user(user)
return {
"refresh": str(refresh),
"access": str(refresh.access_token),
}
+21 -145
View File
@@ -1,25 +1,24 @@
"""API endpoints"""
from django.contrib.postgres.search import TrigramSimilarity
from django.db.models import Func, Max, OuterRef, Prefetch, Q, Subquery, Value
from django.db.models.functions import Coalesce
from django.db.models import Func, Max, OuterRef, Q, Subquery, Value
from rest_framework import (
decorators,
exceptions,
filters,
mixins,
pagination,
response,
throttling,
viewsets,
)
from rest_framework.throttling import UserRateThrottle
from core import models
from . import permissions, serializers
SIMILARITY_THRESHOLD = 0.04
EMAIL_SIMILARITY_THRESHOLD = 0.01
# TrigramSimilarity threshold is lower for searching email than for names,
# to improve matching results
class NestedGenericViewSet(viewsets.GenericViewSet):
@@ -97,11 +96,12 @@ class SerializerPerActionMixin:
class Pagination(pagination.PageNumberPagination):
"""Pagination to display no more than 100 objects per page sorted by creation date."""
ordering = "-created_on"
max_page_size = 100
page_size_query_param = "page_size"
class BurstRateThrottle(throttling.UserRateThrottle):
class BurstRateThrottle(UserRateThrottle):
"""
Throttle rate for minutes. See DRF section in settings for default value.
"""
@@ -109,7 +109,7 @@ class BurstRateThrottle(throttling.UserRateThrottle):
scope = "burst"
class SustainedRateThrottle(throttling.UserRateThrottle):
class SustainedRateThrottle(UserRateThrottle):
"""
Throttle rate for hours. See DRF section in settings for default value.
"""
@@ -185,7 +185,7 @@ class UserViewSet(
"""
permission_classes = [permissions.IsSelf]
queryset = models.User.objects.all().order_by("-created_at")
queryset = models.User.objects.all().select_related("profile_contact")
serializer_class = serializers.UserSerializer
throttle_classes = [BurstRateThrottle, SustainedRateThrottle]
pagination_class = Pagination
@@ -198,37 +198,19 @@ class UserViewSet(
# Exclude inactive contacts
queryset = queryset.filter(
is_active=True,
).prefetch_related(
Prefetch(
"identities",
queryset=models.Identity.objects.filter(is_main=True),
to_attr="_identities_main",
)
)
# Exclude all users already in the given team
if team_id := self.request.GET.get("team_id", ""):
queryset = queryset.exclude(teams__id=team_id)
# Search by case-insensitive and accent-insensitive trigram similarity
if query := self.request.GET.get("q", ""):
similarity = Max(
TrigramSimilarity(
Coalesce(
Func("identities__email", function="unaccent"), Value("")
),
Func(Value(query), function="unaccent"),
)
+ TrigramSimilarity(
Coalesce(
Func("identities__name", function="unaccent"), Value("")
),
Func("identities__email", function="unaccent"),
Func(Value(query), function="unaccent"),
)
)
queryset = (
queryset.annotate(similarity=similarity)
.filter(similarity__gte=SIMILARITY_THRESHOLD)
.filter(similarity__gte=EMAIL_SIMILARITY_THRESHOLD)
.order_by("-similarity")
)
@@ -244,9 +226,9 @@ class UserViewSet(
"""
Return information on currently logged user
"""
user = request.user
context = {"request": request}
return response.Response(
self.serializer_class(user, context={"request": request}).data
self.serializer_class(request.user, context=context).data
)
@@ -262,9 +244,6 @@ class TeamViewSet(
permission_classes = [permissions.AccessPermission]
serializer_class = serializers.TeamSerializer
filter_backends = [filters.OrderingFilter]
ordering_fields = ["created_at"]
ordering = ["-created_at"]
queryset = models.Team.objects.all()
def get_queryset(self):
@@ -321,15 +300,8 @@ class TeamAccessViewSet(
lookup_field = "pk"
pagination_class = Pagination
permission_classes = [permissions.AccessPermission]
queryset = (
models.TeamAccess.objects.all().select_related("user").order_by("-created_at")
)
list_serializer_class = serializers.TeamAccessReadOnlySerializer
detail_serializer_class = serializers.TeamAccessSerializer
filter_backends = [filters.OrderingFilter]
ordering = ["role"]
ordering_fields = ["role", "email", "name"]
queryset = models.TeamAccess.objects.all().select_related("user")
serializer_class = serializers.TeamAccessSerializer
def get_permissions(self):
"""User only needs to be authenticated to list team accesses"""
@@ -346,45 +318,23 @@ class TeamAccessViewSet(
context["team_id"] = self.kwargs["team_id"]
return context
def get_serializer_class(self):
if self.action in {"list", "retrieve"}:
return self.list_serializer_class
return self.detail_serializer_class
def get_queryset(self):
"""Return the queryset according to the action."""
queryset = super().get_queryset()
queryset = queryset.filter(team=self.kwargs["team_id"])
if self.action in {"list", "retrieve"}:
# Determine which role the logged-in user has in the team
if self.action == "list":
# Limit to team access instances related to a team THAT also has a team access
# instance for the logged-in user (we don't want to list only the team access
# instances pointing to the logged-in user)
user_role_query = models.TeamAccess.objects.filter(
user=self.request.user, team=self.kwargs["team_id"]
team__accesses__user=self.request.user
).values("role")[:1]
user_main_identity_query = models.Identity.objects.filter(
user=OuterRef("user_id"), is_main=True
)
queryset = (
# The logged-in user should be part of a team to see its accesses
queryset.filter(
team__accesses__user=self.request.user,
)
.prefetch_related(
Prefetch(
"user__identities",
queryset=models.Identity.objects.filter(is_main=True),
to_attr="_identities_main",
)
)
# Abilities are computed based on logged-in user's role and
# the user role on each team access
.annotate(
user_role=Subquery(user_role_query),
email=Subquery(user_main_identity_query.values("email")[:1]),
name=Subquery(user_main_identity_query.values("name")[:1]),
)
.annotate(user_role=Subquery(user_role_query))
.distinct()
)
return queryset
@@ -422,77 +372,3 @@ class TeamAccessViewSet(
raise exceptions.ValidationError({"role": message})
serializer.save()
class InvitationViewset(
mixins.CreateModelMixin,
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.DestroyModelMixin,
viewsets.GenericViewSet,
):
"""API ViewSet for user invitations to team.
GET /api/v1.0/teams/<team_id>/invitations/:<invitation_id>/
Return list of invitations related to that team or or one
team access if an id is provided.
POST /api/v1.0/teams/<team_id>/invitations/ with expected data:
- email: str
- role: str [owner|admin|member]
- issuer : User, automatically added from user making query, if allowed
- team : Team, automatically added from requested URI
Return newly created invitation
PUT / PATCH : Not permitted. Instead of updating your invitation,
delete and create a new one.
DELETE /api/v1.0/teams/<team_id>/invitations/<invitation_id>/
Delete targeted invitation
"""
lookup_field = "id"
pagination_class = Pagination
permission_classes = [permissions.AccessPermission]
queryset = (
models.Invitation.objects.all().select_related("team").order_by("-created_at")
)
serializer_class = serializers.InvitationSerializer
def get_permissions(self):
"""User only needs to be authenticated to list invitations"""
if self.action == "list":
permission_classes = [permissions.IsAuthenticated]
else:
return super().get_permissions()
return [permission() for permission in permission_classes]
def get_serializer_context(self):
"""Extra context provided to the serializer class."""
context = super().get_serializer_context()
context["team_id"] = self.kwargs["team_id"]
return context
def get_queryset(self):
"""Return the queryset according to the action."""
queryset = super().get_queryset()
queryset = queryset.filter(team=self.kwargs["team_id"])
if self.action == "list":
# Determine which role the logged-in user has in the team
user_role_query = models.TeamAccess.objects.filter(
user=self.request.user, team=self.kwargs["team_id"]
).values("role")[:1]
queryset = (
# The logged-in user should be part of a team to see its accesses
queryset.filter(
team__accesses__user=self.request.user,
)
# Abilities are computed based on logged-in user's role and
# the user role on each team access
.annotate(user_role=Subquery(user_role_query))
.distinct()
)
return queryset
@@ -1,7 +1,7 @@
"""Authentication Backends for the People core app."""
"""Authentication for the People core app."""
import logging
from django.conf import settings
from django.core.exceptions import SuspiciousOperation
from django.db import models
from django.utils.translation import gettext_lazy as _
@@ -9,15 +9,18 @@ import requests
from mozilla_django_oidc.auth import (
OIDCAuthenticationBackend as MozillaOIDCAuthenticationBackend,
)
from rest_framework_simplejwt.exceptions import InvalidToken
from core.models import Identity
from .models import Identity
LOGGER = logging.getLogger(__name__)
class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
"""Custom OpenID Connect (OIDC) Authentication Backend.
This class overrides the default OIDC Authentication Backend to accommodate differences
in the User and Identity models, and handles signed and/or encrypted UserInfo response.
in the User and Identity models, as well as the userinfo endpoint behavior in Agent Connect.
"""
def get_userinfo(self, access_token, id_token, payload):
@@ -31,9 +34,9 @@ class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
Note: The id_token and payload parameters are unused in this implementation,
but were kept to preserve base method signature.
Note: It handles signed and/or encrypted UserInfo Response. It is required by
Agent Connect, which follows the OIDC standard. It forces us to override the
base method, which deal with 'application/json' response.
Note: Userinfo endpoint returns a signed JWT, and not 'application/json'. This
constraint comes from Agent Connect. It forces us to override the base method,
which deal with 'application/json' response.
Returns:
- dict: User details dictionary obtained from the OpenID Connect user endpoint.
@@ -67,39 +70,24 @@ class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
user_info = self.get_userinfo(access_token, id_token, payload)
# Compute user name from OIDC name fields as defined in settings
names_list = [
user_info[field]
for field in settings.USER_OIDC_FIELDS_TO_NAME
if user_info.get(field)
]
user_info["name"] = " ".join(names_list) or None
email = user_info.get("email")
sub = user_info.get("sub")
if sub is None:
raise SuspiciousOperation(
raise InvalidToken(
_("User info contained no recognizable user identification")
)
user = (
self.UserModel.objects.filter(identities__sub=sub)
.annotate(
identity_email=models.F("identities__email"),
identity_name=models.F("identities__name"),
)
.annotate(identity_email=models.F("identities__email"))
.distinct()
.first()
)
if user:
email = user_info.get("email")
name = user_info.get("name")
if (
email
and email != user.identity_email
or name
and name != user.identity_name
):
Identity.objects.filter(sub=sub).update(email=email, name=name)
if email and email != user.identity_email:
Identity.objects.filter(sub=sub).update(email=email)
elif self.get_settings("OIDC_CREATE_USER", True):
user = self.create_user(user_info)
@@ -108,15 +96,16 @@ class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
def create_user(self, claims):
"""Return a newly created User instance."""
email = claims.get("email")
sub = claims.get("sub")
if sub is None:
raise SuspiciousOperation(
raise InvalidToken(
_("Claims contained no recognizable user identification")
)
user = self.UserModel.objects.create(password="!") # noqa: S106
Identity.objects.create(
user=user, sub=sub, email=claims.get("email"), name=claims.get("name")
)
user = self.UserModel.objects.create(password="!", email=email) # noqa: S106
Identity.objects.create(user=user, sub=sub, email=email)
return user
-18
View File
@@ -1,18 +0,0 @@
"""Authentication URLs for the People core app."""
from django.urls import path
from mozilla_django_oidc.urls import urlpatterns as mozzila_oidc_urls
from .views import OIDCLogoutCallbackView, OIDCLogoutView
urlpatterns = [
# Override the default 'logout/' path from Mozilla Django OIDC with our custom view.
path("logout/", OIDCLogoutView.as_view(), name="oidc_logout_custom"),
path(
"logout-callback/",
OIDCLogoutCallbackView.as_view(),
name="oidc_logout_callback",
),
*mozzila_oidc_urls,
]
-137
View File
@@ -1,137 +0,0 @@
"""Authentication Views for the People core app."""
from urllib.parse import urlencode
from django.contrib import auth
from django.core.exceptions import SuspiciousOperation
from django.http import HttpResponseRedirect
from django.urls import reverse
from django.utils import crypto
from mozilla_django_oidc.utils import (
absolutify,
)
from mozilla_django_oidc.views import (
OIDCLogoutView as MozillaOIDCOIDCLogoutView,
)
class OIDCLogoutView(MozillaOIDCOIDCLogoutView):
"""Custom logout view for handling OpenID Connect (OIDC) logout flow.
Adds support for handling logout callbacks from the identity provider (OP)
by initiating the logout flow if the user has an active session.
The Django session is retained during the logout process to persist the 'state' OIDC parameter.
This parameter is crucial for maintaining the integrity of the logout flow between this call
and the subsequent callback.
"""
@staticmethod
def persist_state(request, state):
"""Persist the given 'state' parameter in the session's 'oidc_states' dictionary
This method is used to store the OIDC state parameter in the session, according to the
structure expected by Mozilla Django OIDC's 'add_state_and_verifier_and_nonce_to_session'
utility function.
"""
if "oidc_states" not in request.session or not isinstance(
request.session["oidc_states"], dict
):
request.session["oidc_states"] = {}
request.session["oidc_states"][state] = {}
request.session.save()
def construct_oidc_logout_url(self, request):
"""Create the redirect URL for interfacing with the OIDC provider.
Retrieves the necessary parameters from the session and constructs the URL
required to initiate logout with the OpenID Connect provider.
If no ID token is found in the session, the logout flow will not be initiated,
and the method will return the default redirect URL.
The 'state' parameter is generated randomly and persisted in the session to ensure
its integrity during the subsequent callback.
"""
oidc_logout_endpoint = self.get_settings("OIDC_OP_LOGOUT_ENDPOINT")
if not oidc_logout_endpoint:
return self.redirect_url
reverse_url = reverse("oidc_logout_callback")
id_token = request.session.get("oidc_id_token", None)
if not id_token:
return self.redirect_url
query = {
"id_token_hint": id_token,
"state": crypto.get_random_string(self.get_settings("OIDC_STATE_SIZE", 32)),
"post_logout_redirect_uri": absolutify(request, reverse_url),
}
self.persist_state(request, query["state"])
return f"{oidc_logout_endpoint}?{urlencode(query)}"
def post(self, request):
"""Handle user logout.
If the user is not authenticated, redirects to the default logout URL.
Otherwise, constructs the OIDC logout URL and redirects the user to start
the logout process.
If the user is redirected to the default logout URL, ensure her Django session
is terminated.
"""
logout_url = self.redirect_url
if request.user.is_authenticated:
logout_url = self.construct_oidc_logout_url(request)
# If the user is not redirected to the OIDC provider, ensure logout
if logout_url == self.redirect_url:
auth.logout(request)
return HttpResponseRedirect(logout_url)
class OIDCLogoutCallbackView(MozillaOIDCOIDCLogoutView):
"""Custom view for handling the logout callback from the OpenID Connect (OIDC) provider.
Handles the callback after logout from the identity provider (OP).
Verifies the state parameter and performs necessary logout actions.
The Django session is maintained during the logout process to ensure the integrity
of the logout flow initiated in the previous step.
"""
http_method_names = ["get"]
def get(self, request):
"""Handle the logout callback.
If the user is not authenticated, redirects to the default logout URL.
Otherwise, verifies the state parameter and performs necessary logout actions.
"""
if not request.user.is_authenticated:
return HttpResponseRedirect(self.redirect_url)
state = request.GET.get("state")
if state not in request.session.get("oidc_states", {}):
msg = "OIDC callback state not found in session `oidc_states`!"
raise SuspiciousOperation(msg)
del request.session["oidc_states"][state]
request.session.save()
auth.logout(request)
return HttpResponseRedirect(self.redirect_url)
-10
View File
@@ -1,9 +1,7 @@
"""
Core application enums declaration
"""
from django.conf import global_settings, settings
from django.db import models
from django.utils.translation import gettext_lazy as _
# Django sets `LANGUAGES` by default with all supported languages. We can use it for
@@ -15,11 +13,3 @@ ALL_LANGUAGES = getattr(
"ALL_LANGUAGES",
[(language, _(name)) for language, name in global_settings.LANGUAGES],
)
class WebhookStatusChoices(models.TextChoices):
"""Defines the possible statuses in which a webhook can be."""
FAILURE = "failure", _("Failure")
PENDING = "pending", _("Pending")
SUCCESS = "success", _("Success")
+8 -35
View File
@@ -2,7 +2,6 @@
"""
Core application factories
"""
from django.conf import settings
from django.contrib.auth.hashers import make_password
@@ -120,13 +119,12 @@ class ContactFactory(BaseContactFactory):
class UserFactory(factory.django.DjangoModelFactory):
"""A factory to create random users for testing purposes."""
"""A factory to random users for testing purposes."""
class Meta:
model = models.User
django_get_or_create = ("admin_email",)
admin_email = factory.Faker("email")
email = factory.Faker("email")
language = factory.fuzzy.FuzzyChoice([lang[0] for lang in settings.LANGUAGES])
password = make_password("password")
@@ -141,7 +139,6 @@ class IdentityFactory(factory.django.DjangoModelFactory):
user = factory.SubFactory(UserFactory)
sub = factory.Sequence(lambda n: f"user{n!s}")
email = factory.Faker("email")
name = factory.Faker("name")
class TeamFactory(factory.django.DjangoModelFactory):
@@ -150,20 +147,18 @@ class TeamFactory(factory.django.DjangoModelFactory):
class Meta:
model = models.Team
django_get_or_create = ("name",)
skip_postgeneration_save = True
name = factory.Sequence(lambda n: f"team{n}")
@factory.post_generation
def users(self, create, extracted, **kwargs):
"""Add users to team from a given list of users with or without roles."""
if not create or not extracted:
return
for user_entry in extracted:
if isinstance(user_entry, models.User):
TeamAccessFactory(team=self, user=user_entry)
else:
TeamAccessFactory(team=self, user=user_entry[0], role=user_entry[1])
if create and extracted:
for item in extracted:
if isinstance(item, models.User):
TeamAccessFactory(team=self, user=item)
else:
TeamAccessFactory(team=self, user=item[0], role=item[1])
class TeamAccessFactory(factory.django.DjangoModelFactory):
@@ -175,25 +170,3 @@ class TeamAccessFactory(factory.django.DjangoModelFactory):
team = factory.SubFactory(TeamFactory)
user = factory.SubFactory(UserFactory)
role = factory.fuzzy.FuzzyChoice([r[0] for r in models.RoleChoices.choices])
class TeamWebhookFactory(factory.django.DjangoModelFactory):
"""Create fake team webhooks for testing."""
class Meta:
model = models.TeamWebhook
team = factory.SubFactory(TeamFactory)
url = factory.Sequence(lambda n: f"https://example.com/Groups/{n!s}")
class InvitationFactory(factory.django.DjangoModelFactory):
"""A factory to create invitations for a user"""
class Meta:
model = models.Invitation
team = factory.SubFactory(TeamFactory)
email = factory.Faker("email")
role = factory.fuzzy.FuzzyChoice([role[0] for role in models.RoleChoices.choices])
issuer = factory.SubFactory(UserFactory)
+5 -44
View File
@@ -1,6 +1,6 @@
# Generated by Django 5.0.3 on 2024-03-25 22:58
# Generated by Django 5.0.1 on 2024-02-06 15:08
import django.contrib.auth.models
import core.models
import django.core.validators
import django.db.models.deletion
import timezone_field.fields
@@ -27,7 +27,7 @@ class Migration(migrations.Migration):
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('name', models.CharField(max_length=100)),
('slug', models.SlugField(editable=False, max_length=100, unique=True)),
('slug', models.SlugField(max_length=100, unique=True)),
],
options={
'verbose_name': 'Team',
@@ -45,7 +45,7 @@ class Migration(migrations.Migration):
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('admin_email', models.EmailField(blank=True, max_length=254, null=True, unique=True, verbose_name='admin email address')),
('email', models.EmailField(blank=True, max_length=254, null=True, unique=True, verbose_name='email address')),
('language', models.CharField(choices="(('en-us', 'English'), ('fr-fr', 'French'))", default='en-us', help_text='The language in which the user wants to see the interface.', max_length=10, verbose_name='language')),
('timezone', timezone_field.fields.TimeZoneField(choices_display='WITH_GMT_OFFSET', default='UTC', help_text='The timezone in which the user wants to see times.', use_pytz=False)),
('is_device', models.BooleanField(default=False, help_text='Whether the user is a device or a real user.', verbose_name='device')),
@@ -60,7 +60,7 @@ class Migration(migrations.Migration):
'db_table': 'people_user',
},
managers=[
('objects', django.contrib.auth.models.UserManager()),
('objects', core.models.UserManager()),
],
),
migrations.CreateModel(
@@ -95,7 +95,6 @@ class Migration(migrations.Migration):
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('sub', models.CharField(help_text='Required. 255 characters or fewer. Letters, numbers, and @/./+/-/_ characters only.', max_length=255, unique=True, validators=[django.core.validators.RegexValidator(message='Enter a valid sub. This value may contain only letters, numbers, and @/./+/-/_ characters.', regex='^[\\w.@+-]+\\Z')], verbose_name='sub')),
('email', models.EmailField(blank=True, max_length=254, null=True, verbose_name='email address')),
('name', models.CharField(blank=True, max_length=100, null=True, verbose_name='name')),
('is_main', models.BooleanField(default=False, help_text='Designates whether the email is the main one.', verbose_name='main')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='identities', to=settings.AUTH_USER_MODEL)),
],
@@ -106,23 +105,6 @@ class Migration(migrations.Migration):
'ordering': ('-is_main', 'email'),
},
),
migrations.CreateModel(
name='Invitation',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('email', models.EmailField(max_length=254, verbose_name='email address')),
('role', models.CharField(choices=[('member', 'Member'), ('administrator', 'Administrator'), ('owner', 'Owner')], default='member', max_length=20)),
('issuer', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='invitations', to=settings.AUTH_USER_MODEL)),
('team', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='invitations', to='core.team')),
],
options={
'verbose_name': 'Team invitation',
'verbose_name_plural': 'Team invitations',
'db_table': 'people_invitation',
},
),
migrations.CreateModel(
name='TeamAccess',
fields=[
@@ -144,23 +126,6 @@ class Migration(migrations.Migration):
name='users',
field=models.ManyToManyField(related_name='teams', through='core.TeamAccess', to=settings.AUTH_USER_MODEL),
),
migrations.CreateModel(
name='TeamWebhook',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('url', models.URLField(verbose_name='url')),
('secret', models.CharField(blank=True, max_length=255, null=True, verbose_name='secret')),
('status', models.CharField(choices=[('failure', 'Failure'), ('pending', 'Pending'), ('success', 'Success')], default='pending', max_length=10)),
('team', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='webhooks', to='core.team')),
],
options={
'verbose_name': 'Team webhook',
'verbose_name_plural': 'Team webhooks',
'db_table': 'people_team_webhook',
},
),
migrations.AddConstraint(
model_name='contact',
constraint=models.CheckConstraint(check=models.Q(('base__isnull', False), ('owner__isnull', True), _negated=True), name='base_owner_constraint', violation_error_message='A contact overriding a base contact must be owned.'),
@@ -177,10 +142,6 @@ class Migration(migrations.Migration):
model_name='identity',
constraint=models.UniqueConstraint(fields=('user', 'email'), name='unique_user_email', violation_error_message='This email address is already declared for this user.'),
),
migrations.AddConstraint(
model_name='invitation',
constraint=models.UniqueConstraint(fields=('email', 'team'), name='email_and_team_unique_together'),
),
migrations.AddConstraint(
model_name='teamaccess',
constraint=models.UniqueConstraint(fields=('user', 'team'), name='unique_team_user', violation_error_message='This user is already in this team.'),
+22 -250
View File
@@ -1,35 +1,22 @@
"""
Declare and configure the models for the People core application
"""
import json
import os
import smtplib
import uuid
from datetime import timedelta
from logging import getLogger
from django.conf import settings
from django.contrib.auth import models as auth_models
from django.contrib.auth.base_user import AbstractBaseUser
from django.contrib.sites.models import Site
from django.core import exceptions, mail, validators
from django.db import models, transaction
from django.template.loader import render_to_string
from django.utils import timezone
from django.db import models
from django.utils.functional import lazy
from django.utils.text import slugify
from django.utils.translation import gettext_lazy as _
from django.utils.translation import override
import jsonschema
from timezone_field import TimeZoneField
from core.enums import WebhookStatusChoices
from core.utils.webhooks import scim_synchronizer
logger = getLogger(__name__)
current_dir = os.path.dirname(os.path.abspath(__file__))
contact_schema_path = os.path.join(current_dir, "jsonschema", "contact_data.json")
with open(contact_schema_path, "r", encoding="utf-8") as contact_schema_file:
@@ -78,7 +65,7 @@ class BaseModel(models.Model):
def save(self, *args, **kwargs):
"""Call `full_clean` before saving."""
self.full_clean()
return super().save(*args, **kwargs)
super().save(*args, **kwargs)
class Contact(BaseModel):
@@ -158,12 +145,20 @@ class Contact(BaseModel):
raise exceptions.ValidationError({"data": [error_message]}) from e
class UserManager(auth_models.UserManager):
"""
Override user manager to get the related contact in the same query by default (Contact model)
"""
def get_queryset(self):
"""Always select the related contact when doing a query on users."""
return super().get_queryset().select_related("profile_contact")
class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
"""User model to work with OIDC only authentication."""
admin_email = models.EmailField(
_("admin email address"), unique=True, null=True, blank=True
)
email = models.EmailField(_("email address"), unique=True, null=True, blank=True)
profile_contact = models.OneToOneField(
Contact,
on_delete=models.SET_NULL,
@@ -203,9 +198,9 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
),
)
objects = auth_models.UserManager()
objects = UserManager()
USERNAME_FIELD = "admin_email"
USERNAME_FIELD = "email"
REQUIRED_FIELDS = []
class Meta:
@@ -217,32 +212,9 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
return (
str(self.profile_contact)
if self.profile_contact
else self.admin_email or str(self.id)
else self.email or str(self.id)
)
def _get_identities_main(self):
"""Return a list with the main identity or an empty list."""
try:
return self._identities_main
except AttributeError:
return self.identities.filter(is_main=True)
@property
def name(self):
"""Return main identity's name."""
try:
return self._get_identities_main()[0].name
except IndexError:
return None
@property
def email(self):
"""Return main identity's email."""
try:
return self._get_identities_main()[0].email
except IndexError:
return None
def clean(self):
"""Validate fields."""
super().clean()
@@ -254,10 +226,8 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
def email_user(self, subject, message, from_email=None, **kwargs):
"""Email this user."""
email = self.email or self.admin_email
if not email:
raise ValueError("You must first set an email for the user.")
mail.send_mail(subject, message, from_email, [email], **kwargs)
main_identity = self.identities.get(is_main=True)
mail.send_mail(subject, message, from_email, [main_identity.email], **kwargs)
@classmethod
def get_email_field_name(cls):
@@ -292,7 +262,6 @@ class Identity(BaseModel):
validators=[sub_validator],
)
email = models.EmailField(_("email address"), null=True, blank=True)
name = models.CharField(_("name"), max_length=100, null=True, blank=True)
is_main = models.BooleanField(
_("main"),
default=False,
@@ -317,49 +286,14 @@ class Identity(BaseModel):
def __str__(self):
main_str = "[main]" if self.is_main else ""
id_str = self.email or self.sub
return f"{id_str:s}{main_str:s}"
return f"{self.email:s}{main_str:s}"
def save(self, *args, **kwargs):
"""
Saves identity, ensuring users always have exactly one main identity.
If it's a new identity, give its user access to the relevant teams.
"""
if self._state.adding:
self._convert_valid_invitations()
"""Ensure users always have one and only one main identity."""
super().save(*args, **kwargs)
# Ensure users always have one and only one main identity.
if self.is_main is True:
self.user.identities.exclude(id=self.id).update(is_main=False)
def _convert_valid_invitations(self):
"""
Convert valid invitations to team accesses.
Expired invitations are ignored.
"""
valid_invitations = Invitation.objects.filter(
email=self.email,
created_at__gte=(
timezone.now()
- timedelta(seconds=settings.INVITATION_VALIDITY_DURATION)
),
).select_related("team")
if not valid_invitations.exists():
return
TeamAccess.objects.bulk_create(
[
TeamAccess(user=self.user, team=invitation.team, role=invitation.role)
for invitation in valid_invitations
]
)
valid_invitations.delete()
def clean(self):
"""Normalize the email field and clean the 'is_main' field."""
if self.email:
@@ -399,7 +333,7 @@ class Team(BaseModel):
return self.name
def save(self, *args, **kwargs):
"""Override save function to compute the slug."""
"""Overriding save function to compute the slug."""
self.slug = self.get_slug()
return super().save(*args, **kwargs)
@@ -426,7 +360,7 @@ class Team(BaseModel):
is_owner_or_admin = role in [RoleChoices.OWNER, RoleChoices.ADMIN]
return {
"get": bool(role),
"get": True,
"patch": is_owner_or_admin,
"put": is_owner_or_admin,
"delete": role == RoleChoices.OWNER,
@@ -466,34 +400,6 @@ class TeamAccess(BaseModel):
def __str__(self):
return f"{self.user!s} is {self.role:s} in team {self.team!s}"
def save(self, *args, **kwargs):
"""
Override save function to fire webhooks on any addition or update
to a team access.
"""
if self._state.adding:
self.team.webhooks.update(status=WebhookStatusChoices.PENDING)
with transaction.atomic():
instance = super().save(*args, **kwargs)
scim_synchronizer.add_user_to_group(self.team, self.user)
else:
instance = super().save(*args, **kwargs)
return instance
def delete(self, *args, **kwargs):
"""
Override delete method to fire webhooks on to team accesses.
Don't allow deleting a team access until it is successfully synchronized with all
its webhooks.
"""
self.team.webhooks.update(status=WebhookStatusChoices.PENDING)
with transaction.atomic():
arguments = self.team, self.user
super().delete(*args, **kwargs)
scim_synchronizer.remove_user_from_group(*arguments)
def get_abilities(self, user):
"""
Compute and return abilities for a given user taking into account
@@ -542,137 +448,3 @@ class TeamAccess(BaseModel):
"put": bool(set_role_to),
"set_role_to": set_role_to,
}
class TeamWebhook(BaseModel):
"""Webhooks fired on changes in teams."""
team = models.ForeignKey(Team, related_name="webhooks", on_delete=models.CASCADE)
url = models.URLField(_("url"))
secret = models.CharField(_("secret"), max_length=255, null=True, blank=True)
status = models.CharField(
max_length=10,
default=WebhookStatusChoices.PENDING,
choices=WebhookStatusChoices.choices,
)
class Meta:
db_table = "people_team_webhook"
verbose_name = _("Team webhook")
verbose_name_plural = _("Team webhooks")
def __str__(self):
return f"Webhook to {self.url} for {self.team}"
def get_headers(self):
"""Build header dict from webhook object."""
headers = {"Content-Type": "application/json"}
if self.secret:
headers["Authorization"] = f"Bearer {self.secret:s}"
return headers
class Invitation(BaseModel):
"""User invitation to teams."""
email = models.EmailField(_("email address"), null=False, blank=False)
team = models.ForeignKey(
Team,
on_delete=models.CASCADE,
related_name="invitations",
)
role = models.CharField(
max_length=20, choices=RoleChoices.choices, default=RoleChoices.MEMBER
)
issuer = models.ForeignKey(
User,
on_delete=models.CASCADE,
related_name="invitations",
)
class Meta:
db_table = "people_invitation"
verbose_name = _("Team invitation")
verbose_name_plural = _("Team invitations")
constraints = [
models.UniqueConstraint(
fields=["email", "team"], name="email_and_team_unique_together"
)
]
def __str__(self):
return f"{self.email} invited to {self.team}"
def save(self, *args, **kwargs):
"""Make invitations read-only."""
if self.created_at:
raise exceptions.PermissionDenied()
super().save(*args, **kwargs)
self.email_invitation()
def clean(self):
"""Validate fields."""
super().clean()
# Check if an identity already exists for the provided email
if Identity.objects.filter(email=self.email).exists():
raise exceptions.ValidationError(
{"email": _("This email is already associated to a registered user.")}
)
@property
def is_expired(self):
"""Calculate if invitation is still valid or has expired."""
if not self.created_at:
return None
validity_duration = timedelta(seconds=settings.INVITATION_VALIDITY_DURATION)
return timezone.now() > (self.created_at + validity_duration)
def get_abilities(self, user):
"""Compute and return abilities for a given user."""
can_delete = False
role = None
if user.is_authenticated:
try:
role = self.user_role
except AttributeError:
try:
role = self.team.accesses.filter(user=user).values("role")[0][
"role"
]
except (self._meta.model.DoesNotExist, IndexError):
role = None
can_delete = role in [RoleChoices.OWNER, RoleChoices.ADMIN]
return {
"delete": can_delete,
"get": bool(role),
"patch": False,
"put": False,
}
def email_invitation(self):
"""Email invitation to the user."""
try:
with override(self.issuer.language):
template_vars = {
"title": _("Invitation to join Desk!"),
"site": Site.objects.get_current(),
}
msg_html = render_to_string("mail/html/invitation.html", template_vars)
msg_plain = render_to_string("mail/text/invitation.txt", template_vars)
mail.send_mail(
_("Invitation to join Desk!"),
msg_plain,
settings.EMAIL_FROM,
[self.email],
html_message=msg_html,
fail_silently=False,
)
except smtplib.SMTPException as exception:
logger.error("invitation to %s was not sent: %s", self.email, exception)
Binary file not shown.

Before

Width:  |  Height:  |  Size: 13 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 5.5 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 40 KiB

@@ -1,58 +0,0 @@
"""Custom template tags for the core application of People."""
import base64
from django import template
from django.contrib.staticfiles import finders
from PIL import ImageFile as PillowImageFile
register = template.Library()
def image_to_base64(file_or_path, close=False):
"""
Return the src string of the base64 encoding of an image represented by its path
or file opened or not.
Inspired by Django's "get_image_dimensions"
"""
pil_parser = PillowImageFile.Parser()
if hasattr(file_or_path, "read"):
file = file_or_path
if file.closed and hasattr(file, "open"):
file_or_path.open()
file_pos = file.tell()
file.seek(0)
else:
try:
# pylint: disable=consider-using-with
file = open(file_or_path, "rb")
except OSError:
return ""
close = True
try:
image_data = file.read()
if not image_data:
return ""
pil_parser.feed(image_data)
if pil_parser.image:
mime_type = pil_parser.image.get_format_mimetype()
encoded_string = base64.b64encode(image_data)
return f"data:{mime_type:s};base64, {encoded_string.decode('utf-8'):s}"
return ""
finally:
if close:
file.close()
else:
file.seek(file_pos)
@register.simple_tag
def base64_static(path):
"""Return a static file into a base64."""
full_path = finders.find(path)
if full_path:
return image_to_base64(full_path, True)
return ""
@@ -1,10 +0,0 @@
"""Unit tests for the Authentication URLs."""
from core.authentication.urls import urlpatterns
def test_urls_override_default_mozilla_django_oidc():
"""Custom URL patterns should override default ones from Mozilla Django OIDC."""
url_names = [u.name for u in urlpatterns]
assert url_names.index("oidc_logout_custom") < url_names.index("oidc_logout")
@@ -1,238 +0,0 @@
"""Unit tests for the Authentication Views."""
from unittest import mock
from urllib.parse import parse_qs, urlparse
from django.contrib.auth.models import AnonymousUser
from django.contrib.sessions.middleware import SessionMiddleware
from django.core.exceptions import SuspiciousOperation
from django.test import RequestFactory
from django.test.utils import override_settings
from django.urls import reverse
from django.utils import crypto
import pytest
from rest_framework.test import APIClient
from core import factories
from core.authentication.views import OIDCLogoutCallbackView, OIDCLogoutView
pytestmark = pytest.mark.django_db
@override_settings(LOGOUT_REDIRECT_URL="/example-logout")
def test_view_logout_anonymous():
"""Anonymous users calling the logout url,
should be redirected to the specified LOGOUT_REDIRECT_URL."""
url = reverse("oidc_logout_custom")
response = APIClient().get(url)
assert response.status_code == 302
assert response.url == "/example-logout"
@mock.patch.object(
OIDCLogoutView, "construct_oidc_logout_url", return_value="/example-logout"
)
def test_view_logout(mocked_oidc_logout_url):
"""Authenticated users should be redirected to OIDC provider for logout."""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
url = reverse("oidc_logout_custom")
response = client.get(url)
mocked_oidc_logout_url.assert_called_once()
assert response.status_code == 302
assert response.url == "/example-logout"
@override_settings(LOGOUT_REDIRECT_URL="/default-redirect-logout")
@mock.patch.object(
OIDCLogoutView, "construct_oidc_logout_url", return_value="/default-redirect-logout"
)
def test_view_logout_no_oidc_provider(mocked_oidc_logout_url):
"""Authenticated users should be logged out when no OIDC provider is available."""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
url = reverse("oidc_logout_custom")
with mock.patch("mozilla_django_oidc.views.auth.logout") as mock_logout:
response = client.get(url)
mocked_oidc_logout_url.assert_called_once()
mock_logout.assert_called_once()
assert response.status_code == 302
assert response.url == "/default-redirect-logout"
@override_settings(LOGOUT_REDIRECT_URL="/example-logout")
def test_view_logout_callback_anonymous():
"""Anonymous users calling the logout callback url,
should be redirected to the specified LOGOUT_REDIRECT_URL."""
url = reverse("oidc_logout_callback")
response = APIClient().get(url)
assert response.status_code == 302
assert response.url == "/example-logout"
@pytest.mark.parametrize(
"initial_oidc_states",
[{}, {"other_state": "foo"}],
)
def test_view_logout_persist_state(initial_oidc_states):
"""State value should be persisted in session's data."""
identity = factories.IdentityFactory()
user = identity.user
request = RequestFactory().request()
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
if initial_oidc_states:
request.session["oidc_states"] = initial_oidc_states
request.session.save()
mocked_state = "mock_state"
OIDCLogoutView().persist_state(request, mocked_state)
assert "oidc_states" in request.session
assert request.session["oidc_states"] == {
"mock_state": {},
**initial_oidc_states,
}
@override_settings(OIDC_OP_LOGOUT_ENDPOINT="/example-logout")
@mock.patch.object(OIDCLogoutView, "persist_state")
@mock.patch.object(crypto, "get_random_string", return_value="mocked_state")
def test_view_logout_construct_oidc_logout_url(
mocked_get_random_string, mocked_persist_state
):
"""Should construct the logout URL to initiate the logout flow with the OIDC provider."""
identity = factories.IdentityFactory()
user = identity.user
request = RequestFactory().request()
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
request.session["oidc_id_token"] = "mocked_oidc_id_token"
request.session.save()
redirect_url = OIDCLogoutView().construct_oidc_logout_url(request)
mocked_persist_state.assert_called_once()
mocked_get_random_string.assert_called_once()
params = parse_qs(urlparse(redirect_url).query)
assert params["id_token_hint"][0] == "mocked_oidc_id_token"
assert params["state"][0] == "mocked_state"
url = reverse("oidc_logout_callback")
assert url in params["post_logout_redirect_uri"][0]
@override_settings(LOGOUT_REDIRECT_URL="/")
def test_view_logout_construct_oidc_logout_url_none_id_token():
"""If no ID token is available in the session,
the user should be redirected to the final URL."""
identity = factories.IdentityFactory()
user = identity.user
request = RequestFactory().request()
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
redirect_url = OIDCLogoutView().construct_oidc_logout_url(request)
assert redirect_url == "/"
@pytest.mark.parametrize(
"initial_state",
[None, {"other_state": "foo"}],
)
def test_view_logout_callback_wrong_state(initial_state):
"""Should raise an error if OIDC state doesn't match session data."""
identity = factories.IdentityFactory()
user = identity.user
request = RequestFactory().request()
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
if initial_state:
request.session["oidc_states"] = initial_state
request.session.save()
callback_view = OIDCLogoutCallbackView.as_view()
with pytest.raises(SuspiciousOperation) as excinfo:
callback_view(request)
assert (
str(excinfo.value) == "OIDC callback state not found in session `oidc_states`!"
)
@override_settings(LOGOUT_REDIRECT_URL="/example-logout")
def test_view_logout_callback():
"""If state matches, callback should clear OIDC state and redirects."""
identity = factories.IdentityFactory()
user = identity.user
request = RequestFactory().get("/logout-callback/", data={"state": "mocked_state"})
request.user = user
middleware = SessionMiddleware(get_response=lambda x: x)
middleware.process_request(request)
mocked_state = "mocked_state"
request.session["oidc_states"] = {mocked_state: {}}
request.session.save()
callback_view = OIDCLogoutCallbackView.as_view()
with mock.patch("mozilla_django_oidc.views.auth.logout") as mock_logout:
def clear_user(request):
# Assert state is cleared prior to logout
assert request.session["oidc_states"] == {}
request.user = AnonymousUser()
mock_logout.side_effect = clear_user
response = callback_view(request)
mock_logout.assert_called_once()
assert response.status_code == 302
assert response.url == "/example-logout"
@@ -1,7 +1,6 @@
"""
Test suite for generated openapi schema.
"""
import json
from io import StringIO
@@ -1,236 +0,0 @@
"""
Test for team accesses API endpoints in People's core app : create
"""
import json
import random
import re
import pytest
import responses
from rest_framework.test import APIClient
from core import factories, models
pytestmark = pytest.mark.django_db
def test_api_team_accesses_create_anonymous():
"""Anonymous users should not be allowed to create team accesses."""
user = factories.UserFactory()
team = factories.TeamFactory()
response = APIClient().post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(user.id),
"team": str(team.id),
"role": random.choice(models.RoleChoices.choices)[0],
},
format="json",
)
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
assert models.TeamAccess.objects.exists() is False
def test_api_team_accesses_create_authenticated_unrelated():
"""
Authenticated users should not be allowed to create team accesses for a team to
which they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
other_user = factories.UserFactory()
team = factories.TeamFactory()
client = APIClient()
client.force_login(user)
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
},
format="json",
)
assert response.status_code == 403
assert response.json() == {
"detail": "You are not allowed to manage accesses for this team."
}
assert not models.TeamAccess.objects.filter(user=other_user).exists()
def test_api_team_accesses_create_authenticated_member():
"""Members of a team should not be allowed to create team accesses."""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "member")])
other_user = factories.UserFactory()
client = APIClient()
client.force_login(user)
for role in [role[0] for role in models.RoleChoices.choices]:
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
)
assert response.status_code == 403
assert response.json() == {
"detail": "You are not allowed to manage accesses for this team."
}
assert not models.TeamAccess.objects.filter(user=other_user).exists()
def test_api_team_accesses_create_authenticated_administrator():
"""
Administrators of a team should be able to create team accesses except for the "owner" role.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "administrator")])
other_user = factories.UserFactory()
client = APIClient()
client.force_login(user)
# It should not be allowed to create an owner access
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": "owner",
},
format="json",
)
assert response.status_code == 403
assert response.json() == {
"detail": "Only owners of a team can assign other users as owners."
}
# It should be allowed to create a lower access
role = random.choice(
[role[0] for role in models.RoleChoices.choices if role[0] != "owner"]
)
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
)
assert response.status_code == 201
assert models.TeamAccess.objects.filter(user=other_user).count() == 1
new_team_access = models.TeamAccess.objects.filter(user=other_user).get()
assert response.json() == {
"abilities": new_team_access.get_abilities(user),
"id": str(new_team_access.id),
"role": role,
"user": str(other_user.id),
}
def test_api_team_accesses_create_authenticated_owner():
"""
Owners of a team should be able to create team accesses whatever the role.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "owner")])
other_user = factories.UserFactory()
role = random.choice([role[0] for role in models.RoleChoices.choices])
client = APIClient()
client.force_login(user)
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
)
assert response.status_code == 201
assert models.TeamAccess.objects.filter(user=other_user).count() == 1
new_team_access = models.TeamAccess.objects.filter(user=other_user).get()
assert response.json() == {
"abilities": new_team_access.get_abilities(user),
"id": str(new_team_access.id),
"role": role,
"user": str(other_user.id),
}
def test_api_team_accesses_create_webhook():
"""
When the team has a webhook, creating a team access should fire a call.
"""
user, other_user = factories.UserFactory.create_batch(2)
team = factories.TeamFactory(users=[(user, "owner")])
webhook = factories.TeamWebhookFactory(team=team)
role = random.choice([role[0] for role in models.RoleChoices.choices])
client = APIClient()
client.force_login(user)
with responses.RequestsMock() as rsps:
# Ensure successful response by scim provider using "responses":
rsp = rsps.add(
rsps.PATCH,
re.compile(r".*/Groups/.*"),
body="{}",
status=200,
content_type="application/json",
)
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
)
assert response.status_code == 201
assert rsp.call_count == 1
assert rsps.calls[0].request.url == webhook.url
# Payload sent to scim provider
payload = json.loads(rsps.calls[0].request.body)
assert payload == {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "add",
"path": "members",
"value": [
{
"value": str(other_user.id),
"email": None,
"type": "User",
}
],
}
],
}
@@ -1,227 +0,0 @@
"""
Test for team accesses API endpoints in People's core app : delete
"""
import json
import random
import re
import pytest
import responses
from rest_framework.test import APIClient
from core import factories, models
pytestmark = pytest.mark.django_db
def test_api_team_accesses_delete_anonymous():
"""Anonymous users should not be allowed to destroy a team access."""
access = factories.TeamAccessFactory()
response = APIClient().delete(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 401
assert models.TeamAccess.objects.count() == 1
def test_api_team_accesses_delete_authenticated():
"""
Authenticated users should not be allowed to delete a team access for a
team to which they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
access = factories.TeamAccessFactory()
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 403
assert models.TeamAccess.objects.count() == 1
def test_api_team_accesses_delete_member():
"""
Authenticated users should not be allowed to delete a team access for a
team in which they are a simple member.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "member")])
access = factories.TeamAccessFactory(team=team)
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 403
assert models.TeamAccess.objects.count() == 2
def test_api_team_accesses_delete_administrators():
"""
Users who are administrators in a team should be allowed to delete an access
from the team provided it is not ownership.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "administrator")])
access = factories.TeamAccessFactory(
team=team, role=random.choice(["member", "administrator"])
)
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 204
assert models.TeamAccess.objects.count() == 1
def test_api_team_accesses_delete_owners_except_owners():
"""
Users should be able to delete the team access of another user
for a team of which they are owner provided it is not an owner access.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "owner")])
access = factories.TeamAccessFactory(
team=team, role=random.choice(["member", "administrator"])
)
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 204
assert models.TeamAccess.objects.count() == 1
def test_api_team_accesses_delete_owners_for_owners():
"""
Users should not be allowed to delete the team access of another owner
even for a team in which they are direct owner.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "owner")])
access = factories.TeamAccessFactory(team=team, role="owner")
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 403
assert models.TeamAccess.objects.count() == 2
def test_api_team_accesses_delete_owners_last_owner():
"""
It should not be possible to delete the last owner access from a team
"""
user = factories.UserFactory()
team = factories.TeamFactory()
access = factories.TeamAccessFactory(team=team, user=user, role="owner")
assert models.TeamAccess.objects.count() == 1
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 403
assert models.TeamAccess.objects.count() == 1
def test_api_team_accesses_delete_webhook():
"""
When the team has a webhook, deleting a team access should fire a call.
"""
user = factories.UserFactory()
team = factories.TeamFactory(users=[(user, "administrator")])
webhook = factories.TeamWebhookFactory(team=team)
access = factories.TeamAccessFactory(
team=team, role=random.choice(["member", "administrator"])
)
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
client = APIClient()
client.force_login(user)
with responses.RequestsMock() as rsps:
# Ensure successful response by scim provider using "responses":
rsp = rsps.add(
rsps.PATCH,
re.compile(r".*/Groups/.*"),
body="{}",
status=200,
content_type="application/json",
)
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 204
assert rsp.call_count == 1
assert rsps.calls[0].request.url == webhook.url
# Payload sent to scim provider
payload = json.loads(rsps.calls[0].request.body)
assert payload == {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "remove",
"path": "members",
"value": [
{
"value": str(access.user.id),
"email": None,
"type": "User",
}
],
}
],
}
assert models.TeamAccess.objects.count() == 1
assert models.TeamAccess.objects.filter(user=access.user).exists() is False
@@ -1,289 +0,0 @@
"""
Test for team accesses API endpoints in People's core app : list
"""
import pytest
from rest_framework.status import HTTP_200_OK
from rest_framework.test import APIClient
from core import factories, models
pytestmark = pytest.mark.django_db
def test_api_team_accesses_list_anonymous():
"""Anonymous users should not be allowed to list team accesses."""
team = factories.TeamFactory()
factories.TeamAccessFactory.create_batch(2, team=team)
response = APIClient().get(f"/api/v1.0/teams/{team.id!s}/accesses/")
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_team_accesses_list_authenticated_unrelated():
"""
Authenticated users should not be allowed to list team accesses for a team
to which they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory()
factories.TeamAccessFactory.create_batch(3, team=team)
# Accesses for other teams to which the user is related should not be listed either
other_access = factories.TeamAccessFactory(user=user)
factories.TeamAccessFactory(team=other_access.team)
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/",
)
assert response.status_code == 200
assert response.json() == {
"count": 0,
"next": None,
"previous": None,
"results": [],
}
def test_api_team_accesses_list_authenticated_related():
"""
Authenticated users should be able to list team accesses for a team
to which they are related, with a given role.
"""
identity = factories.IdentityFactory(is_main=True)
user = identity.user
team = factories.TeamFactory()
owner = factories.IdentityFactory(is_main=True)
access1 = factories.TeamAccessFactory.create(
team=team, user=owner.user, role="owner"
)
administrator = factories.IdentityFactory(is_main=True)
access2 = factories.TeamAccessFactory.create(
team=team, user=administrator.user, role="administrator"
)
# Ensure this user's role is different from other team members to test abilities' computation
user_access = models.TeamAccess.objects.create(team=team, user=user, role="member")
# Grant other team accesses to the user, they should not be listed either
other_access = factories.TeamAccessFactory(user=user)
factories.TeamAccessFactory(team=other_access.team)
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/",
)
assert response.status_code == 200
assert response.json()["count"] == 3
assert sorted(response.json()["results"], key=lambda x: x["id"]) == sorted(
[
{
"id": str(user_access.id),
"user": {
"id": str(user_access.user.id),
"email": str(identity.email),
"name": str(identity.name),
},
"role": str(user_access.role),
"abilities": user_access.get_abilities(user),
},
{
"id": str(access1.id),
"user": {
"id": str(access1.user.id),
"email": str(owner.email),
"name": str(owner.name),
},
"role": str(access1.role),
"abilities": access1.get_abilities(user),
},
{
"id": str(access2.id),
"user": {
"id": str(access2.user.id),
"email": str(administrator.email),
"name": str(administrator.name),
},
"role": str(access2.role),
"abilities": access2.get_abilities(user),
},
],
key=lambda x: x["id"],
)
def test_api_team_accesses_list_authenticated_main_identity():
"""
Name and email should be returned from main identity only
"""
user = factories.UserFactory()
identity = factories.IdentityFactory(user=user, is_main=True)
factories.IdentityFactory(user=user) # additional non-main identity
team = factories.TeamFactory()
models.TeamAccess.objects.create(team=team, user=user) # random role
# other team members should appear, with correct identity
other_user = factories.UserFactory()
other_main_identity = factories.IdentityFactory(is_main=True, user=other_user)
factories.IdentityFactory(user=other_user)
factories.TeamAccessFactory.create(team=team, user=other_user)
# Accesses for other teams to which the user is related should not be listed either
other_access = factories.TeamAccessFactory(user=user)
factories.TeamAccessFactory(team=other_access.team)
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/",
)
assert response.status_code == 200
assert response.json()["count"] == 2
users_info = [
(access["user"]["email"], access["user"]["name"])
for access in response.json()["results"]
]
# user information should be returned from main identity
assert sorted(users_info) == sorted(
[
(str(identity.email), str(identity.name)),
(str(other_main_identity.email), str(other_main_identity.name)),
]
)
def test_api_team_accesses_list_authenticated_constant_numqueries(
django_assert_num_queries,
):
"""
The number of queries should not depend on the amount of fetched accesses.
"""
user = factories.UserFactory()
factories.IdentityFactory(user=user, is_main=True)
team = factories.TeamFactory()
models.TeamAccess.objects.create(team=team, user=user) # random role
client = APIClient()
client.force_login(user)
# Only 4 queries are needed to efficiently fetch team accesses,
# related users and identities :
# - query retrieving logged-in user for user_role annotation
# - count from pagination
# - query prefetching users' main identity
# - distinct from viewset
with django_assert_num_queries(4):
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/",
)
# create 20 new team members
for _ in range(20):
extra_user = factories.IdentityFactory(is_main=True).user
factories.TeamAccessFactory(team=team, user=extra_user)
# num queries should still be 4
with django_assert_num_queries(4):
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/",
)
assert response.status_code == 200
assert response.json()["count"] == 21
def test_api_team_accesses_list_authenticated_ordering():
"""Team accesses can be ordered by "role"."""
user = factories.UserFactory()
factories.IdentityFactory(user=user, is_main=True)
team = factories.TeamFactory()
models.TeamAccess.objects.create(team=team, user=user)
# create 20 new team members
for _ in range(20):
extra_user = factories.IdentityFactory(is_main=True).user
factories.TeamAccessFactory(team=team, user=extra_user)
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/?ordering=role",
)
assert response.status_code == HTTP_200_OK
assert response.json()["count"] == 21
results = [team_access["role"] for team_access in response.json()["results"]]
assert sorted(results) == results
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/?ordering=-role",
)
assert response.status_code == HTTP_200_OK
assert response.json()["count"] == 21
results = [team_access["role"] for team_access in response.json()["results"]]
assert sorted(results, reverse=True) == results
@pytest.mark.parametrize("ordering_fields", ["name", "email"])
def test_api_team_accesses_list_authenticated_ordering_user(ordering_fields):
"""Team accesses can be ordered by user's fields "email" or "name"."""
user = factories.UserFactory()
factories.IdentityFactory(user=user, is_main=True)
team = factories.TeamFactory()
models.TeamAccess.objects.create(team=team, user=user)
# create 20 new team members
for _ in range(20):
extra_user = factories.IdentityFactory(is_main=True).user
factories.TeamAccessFactory(team=team, user=extra_user)
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/?ordering={ordering_fields}",
)
assert response.status_code == HTTP_200_OK
assert response.json()["count"] == 21
def normalize(x):
"""Mimic Django order_by, which is case-insensitive and space-insensitive"""
return x.casefold().replace(" ", "")
results = [
team_access["user"][ordering_fields]
for team_access in response.json()["results"]
]
assert sorted(results, key=normalize) == results
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/?ordering=-{ordering_fields}",
)
assert response.status_code == HTTP_200_OK
assert response.json()["count"] == 21
results = [
team_access["user"][ordering_fields]
for team_access in response.json()["results"]
]
assert sorted(results, reverse=True, key=normalize) == results
@@ -1,87 +0,0 @@
"""
Test for team accesses API endpoints in People's core app : retrieve
"""
import pytest
from rest_framework.test import APIClient
from core import factories
pytestmark = pytest.mark.django_db
def test_api_team_accesses_retrieve_anonymous():
"""
Anonymous users should not be allowed to retrieve a team access.
"""
access = factories.TeamAccessFactory()
response = APIClient().get(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_team_accesses_retrieve_authenticated_unrelated():
"""
Authenticated users should not be allowed to retrieve a team access for
a team to which they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
access = factories.TeamAccessFactory(team=factories.TeamFactory())
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 404
assert response.json() == {"detail": "No TeamAccess matches the given query."}
# Accesses related to another team should be excluded even if the user is related to it
for other_access in [
factories.TeamAccessFactory(),
factories.TeamAccessFactory(user=user),
]:
response = client.get(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{other_access.id!s}/",
)
assert response.status_code == 404
assert response.json() == {"detail": "No TeamAccess matches the given query."}
def test_api_team_accesses_retrieve_authenticated_related():
"""
A user who is related to a team should be allowed to retrieve the
associated team user accesses.
"""
identity = factories.IdentityFactory(is_main=True)
user = identity.user
team = factories.TeamFactory()
access = factories.TeamAccessFactory(team=team, user=user)
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 200
assert response.json() == {
"id": str(access.id),
"user": {
"id": str(access.user.id),
"email": str(identity.email),
"name": str(identity.name),
},
"role": str(access.role),
"abilities": access.get_abilities(user),
}
@@ -1,341 +0,0 @@
"""
Test for team accesses API endpoints in People's core app : update
"""
import random
from uuid import uuid4
import pytest
from rest_framework.test import APIClient
from core import factories, models
from core.api import serializers
pytestmark = pytest.mark.django_db
def test_api_team_accesses_update_anonymous():
"""Anonymous users should not be allowed to update a team access."""
access = factories.TeamAccessFactory()
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
for field, value in new_values.items():
response = APIClient().put(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
{**old_values, field: value},
format="json",
)
assert response.status_code == 401
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_authenticated_unrelated():
"""
Authenticated users should not be allowed to update a team access for a team to which
they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
access = factories.TeamAccessFactory()
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
client = APIClient()
client.force_login(user)
for field, value in new_values.items():
response = client.put(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
{**old_values, field: value},
format="json",
)
assert response.status_code == 403
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_authenticated_member():
"""Members of a team should not be allowed to update its accesses."""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "member")])
access = factories.TeamAccessFactory(team=team)
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
client = APIClient()
client.force_login(user)
for field, value in new_values.items():
response = client.put(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
{**old_values, field: value},
format="json",
)
assert response.status_code == 403
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_administrator_except_owner():
"""
A user who is an administrator in a team should be allowed to update a user
access for this team, as long as they don't try to set the role to owner.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "administrator")])
access = factories.TeamAccessFactory(
team=team,
role=random.choice(["administrator", "member"]),
)
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": random.choice(["administrator", "member"]),
}
client = APIClient()
client.force_login(user)
for field, value in new_values.items():
new_data = {**old_values, field: value}
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data=new_data,
format="json",
)
if (
new_data["role"] == old_values["role"]
): # we are not really updating the role
assert response.status_code == 403
else:
assert response.status_code == 200
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
if field == "role":
assert updated_values == {**old_values, "role": new_values["role"]}
else:
assert updated_values == old_values
def test_api_team_accesses_update_administrator_from_owner():
"""
A user who is an administrator in a team, should not be allowed to update
the user access of an "owner" for this team.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "administrator")])
other_user = factories.UserFactory()
access = factories.TeamAccessFactory(team=team, user=other_user, role="owner")
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
client = APIClient()
client.force_login(user)
for field, value in new_values.items():
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, field: value},
format="json",
)
assert response.status_code == 403
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_administrator_to_owner():
"""
A user who is an administrator in a team, should not be allowed to update
the user access of another user to grant team ownership.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "administrator")])
other_user = factories.UserFactory()
access = factories.TeamAccessFactory(
team=team,
user=other_user,
role=random.choice(["administrator", "member"]),
)
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": "owner",
}
client = APIClient()
client.force_login(user)
for field, value in new_values.items():
new_data = {**old_values, field: value}
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data=new_data,
format="json",
)
# We are not allowed or not really updating the role
if field == "role" or new_data["role"] == old_values["role"]:
assert response.status_code == 403
else:
assert response.status_code == 200
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_owner_except_owner():
"""
A user who is an owner in a team should be allowed to update
a user access for this team except for existing "owner" accesses.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "owner")])
factories.UserFactory()
access = factories.TeamAccessFactory(
team=team,
role=random.choice(["administrator", "member"]),
)
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
client = APIClient()
client.force_login(user)
for field, value in new_values.items():
new_data = {**old_values, field: value}
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data=new_data,
format="json",
)
if (
new_data["role"] == old_values["role"]
): # we are not really updating the role
assert response.status_code == 403
else:
assert response.status_code == 200
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
if field == "role":
assert updated_values == {**old_values, "role": new_values["role"]}
else:
assert updated_values == old_values
def test_api_team_accesses_update_owner_for_owners():
"""
A user who is "owner" of a team should not be allowed to update
an existing owner access for this team.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory(users=[(user, "owner")])
access = factories.TeamAccessFactory(team=team, role="owner")
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
client = APIClient()
client.force_login(user)
for field, value in new_values.items():
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, field: value},
content_type="application/json",
)
assert response.status_code == 403
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_owner_self():
"""
A user who is owner of a team should be allowed to update
their own user access provided there are other owners in the team.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory()
access = factories.TeamAccessFactory(team=team, user=user, role="owner")
old_values = serializers.TeamAccessSerializer(instance=access).data
new_role = random.choice(["administrator", "member"])
client = APIClient()
client.force_login(user)
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, "role": new_role},
format="json",
)
assert response.status_code == 403
access.refresh_from_db()
assert access.role == "owner"
# Add another owner and it should now work
factories.TeamAccessFactory(team=team, role="owner")
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, "role": new_role},
format="json",
)
assert response.status_code == 200
access.refresh_from_db()
assert access.role == new_role
@@ -1,7 +1,6 @@
"""
Tests for Teams API endpoint in People's core app: create
"""
import pytest
from rest_framework.status import (
HTTP_201_CREATED,
@@ -1,7 +1,6 @@
"""
Tests for Teams API endpoint in People's core app: delete
"""
import pytest
from rest_framework.status import (
HTTP_204_NO_CONTENT,
@@ -45,7 +44,7 @@ def test_api_teams_delete_authenticated_unrelated():
)
assert response.status_code == HTTP_404_NOT_FOUND
assert response.json() == {"detail": "No Team matches the given query."}
assert response.json() == {"detail": "Not found."}
assert models.Team.objects.count() == 1
@@ -1,7 +1,6 @@
"""
Tests for Teams API endpoint in People's core app: list
"""
from unittest import mock
import pytest
@@ -27,10 +26,7 @@ def test_api_teams_list_anonymous():
def test_api_teams_list_authenticated():
"""
Authenticated users should be able to list teams
they are an owner/administrator/member of.
"""
"""Authenticated users should be able to list teams they are an owner/administrator/member of."""
identity = factories.IdentityFactory()
user = identity.user
@@ -123,61 +119,3 @@ def test_api_teams_list_authenticated_distinct():
content = response.json()
assert len(content["results"]) == 1
assert content["results"][0]["id"] == str(team.id)
def test_api_teams_order():
"""
Test that the endpoint GET teams is sorted in 'created_at' descending order by default.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team_ids = [
str(team.id) for team in factories.TeamFactory.create_batch(5, users=[user])
]
response = client.get(
"/api/v1.0/teams/",
)
assert response.status_code == 200
response_data = response.json()
response_team_ids = [team["id"] for team in response_data["results"]]
team_ids.reverse()
assert (
response_team_ids == team_ids
), "created_at values are not sorted from newest to oldest"
def test_api_teams_order_param():
"""
Test that the 'created_at' field is sorted in ascending order
when the 'ordering' query parameter is set.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team_ids = [
str(team.id) for team in factories.TeamFactory.create_batch(5, users=[user])
]
response = client.get(
"/api/v1.0/teams/?ordering=created_at",
)
assert response.status_code == 200
response_data = response.json()
response_team_ids = [team["id"] for team in response_data["results"]]
assert (
response_team_ids == team_ids
), "created_at values are not sorted from oldest to newest"
@@ -1,9 +1,8 @@
"""
Tests for Teams API endpoint in People's core app: retrieve
"""
import pytest
from rest_framework import status
from rest_framework.status import HTTP_200_OK, HTTP_401_UNAUTHORIZED, HTTP_404_NOT_FOUND
from rest_framework.test import APIClient
from core import factories
@@ -16,7 +15,7 @@ def test_api_teams_retrieve_anonymous():
team = factories.TeamFactory()
response = APIClient().get(f"/api/v1.0/teams/{team.id}/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert response.status_code == HTTP_401_UNAUTHORIZED
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
@@ -37,8 +36,8 @@ def test_api_teams_retrieve_authenticated_unrelated():
response = client.get(
f"/api/v1.0/teams/{team.id!s}/",
)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json() == {"detail": "No Team matches the given query."}
assert response.status_code == HTTP_404_NOT_FOUND
assert response.json() == {"detail": "Not found."}
def test_api_teams_retrieve_authenticated_related():
@@ -59,19 +58,28 @@ def test_api_teams_retrieve_authenticated_related():
response = client.get(
f"/api/v1.0/teams/{team.id!s}/",
)
assert response.status_code == status.HTTP_200_OK
assert sorted(response.json().pop("accesses")) == sorted(
assert response.status_code == HTTP_200_OK
content = response.json()
assert sorted(content.pop("accesses"), key=lambda x: x["user"]) == sorted(
[
str(access1.id),
str(access2.id),
]
{
"id": str(access1.id),
"user": str(user.id),
"role": access1.role,
"abilities": access1.get_abilities(user),
},
{
"id": str(access2.id),
"user": str(access2.user.id),
"role": access2.role,
"abilities": access2.get_abilities(user),
},
],
key=lambda x: x["user"],
)
assert response.json() == {
"id": str(team.id),
"name": team.name,
"slug": team.slug,
"abilities": team.get_abilities(user),
"created_at": team.created_at.isoformat().replace("+00:00", "Z"),
"updated_at": team.updated_at.isoformat().replace("+00:00", "Z"),
}
@@ -1,7 +1,6 @@
"""
Tests for Teams API endpoint in People's core app: update
"""
import random
import pytest
@@ -61,7 +60,7 @@ def test_api_teams_update_authenticated_unrelated():
)
assert response.status_code == HTTP_404_NOT_FOUND
assert response.json() == {"detail": "No Team matches the given query."}
assert response.json() == {"detail": "Not found."}
team.refresh_from_db()
team_values = serializers.TeamSerializer(instance=team).data
@@ -122,10 +121,8 @@ def test_api_teams_update_authenticated_administrators():
team.refresh_from_db()
final_values = serializers.TeamSerializer(instance=team).data
for key, value in final_values.items():
if key in ["id", "accesses", "created_at"]:
if key in ["id", "accesses"]: # pylint: disable=R1733
assert value == initial_values[key]
elif key == "updated_at":
assert value > initial_values[key]
else:
# name, slug and abilities successfully modified
assert value == new_values[key]
@@ -156,10 +153,8 @@ def test_api_teams_update_authenticated_owners():
team.refresh_from_db()
team_values = serializers.TeamSerializer(instance=team).data
for key, value in team_values.items():
if key in ["id", "accesses", "created_at"]:
if key in ["id", "accesses"]:
assert value == old_team_values[key]
elif key == "updated_at":
assert value > old_team_values[key]
else:
# name, slug and abilities successfully modified
assert value == new_team_values[key]
@@ -188,7 +183,7 @@ def test_api_teams_update_administrator_or_owner_of_another():
)
assert response.status_code == HTTP_404_NOT_FOUND
assert response.json() == {"detail": "No Team matches the given query."}
assert response.json() == {"detail": "Not found."}
team.refresh_from_db()
team_values = serializers.TeamSerializer(instance=team).data
@@ -1,7 +1,6 @@
"""
Test contacts API endpoints in People's core app.
"""
from django.test.utils import override_settings
import pytest
@@ -0,0 +1,846 @@
"""
Test team accesses API endpoints for users in People's core app.
"""
import random
from uuid import uuid4
import pytest
from rest_framework.test import APIClient
from core import factories, models
from core.api import serializers
pytestmark = pytest.mark.django_db
def test_api_team_accesses_list_anonymous():
"""Anonymous users should not be allowed to list team accesses."""
team = factories.TeamFactory()
factories.TeamAccessFactory.create_batch(2, team=team)
response = APIClient().get(f"/api/v1.0/teams/{team.id!s}/accesses/")
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_team_accesses_list_authenticated_unrelated():
"""
Authenticated users should not be allowed to list team accesses for a team
to which they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
team = factories.TeamFactory()
factories.TeamAccessFactory.create_batch(3, team=team)
client = APIClient()
client.force_login(user)
# Accesses for other teams to which the user is related should not be listed either
other_access = factories.TeamAccessFactory(user=user)
factories.TeamAccessFactory(team=other_access.team)
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/",
)
assert response.status_code == 200
assert response.json() == {
"count": 0,
"next": None,
"previous": None,
"results": [],
}
def test_api_team_accesses_list_authenticated_related():
"""
Authenticated users should be able to list team accesses for a team
to which they are related, whatever their role in the team.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory()
user_access = models.TeamAccess.objects.create(team=team, user=user) # random role
access1, access2 = factories.TeamAccessFactory.create_batch(2, team=team)
# Accesses for other teams to which the user is related should not be listed either
other_access = factories.TeamAccessFactory(user=user)
factories.TeamAccessFactory(team=other_access.team)
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/",
)
assert response.status_code == 200
content = response.json()
assert len(content["results"]) == 3
assert sorted(content["results"], key=lambda x: x["id"]) == sorted(
[
{
"id": str(user_access.id),
"user": str(user.id),
"role": user_access.role,
"abilities": user_access.get_abilities(user),
},
{
"id": str(access1.id),
"user": str(access1.user.id),
"role": access1.role,
"abilities": access1.get_abilities(user),
},
{
"id": str(access2.id),
"user": str(access2.user.id),
"role": access2.role,
"abilities": access2.get_abilities(user),
},
],
key=lambda x: x["id"],
)
def test_api_team_accesses_retrieve_anonymous():
"""
Anonymous users should not be allowed to retrieve a team access.
"""
access = factories.TeamAccessFactory()
response = APIClient().get(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_team_accesses_retrieve_authenticated_unrelated():
"""
Authenticated users should not be allowed to retrieve a team access for
a team to which they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory()
access = factories.TeamAccessFactory(team=team)
response = client.get(f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/")
assert response.status_code == 403
assert response.json() == {
"detail": "You do not have permission to perform this action."
}
# Accesses related to another team should be excluded even if the user is related to it
for access in [
factories.TeamAccessFactory(),
factories.TeamAccessFactory(user=user),
]:
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
def test_api_team_accesses_retrieve_authenticated_related():
"""
A user who is related to a team should be allowed to retrieve the
associated team user accesses.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[user])
access = factories.TeamAccessFactory(team=team)
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 200
assert response.json() == {
"id": str(access.id),
"user": str(access.user.id),
"role": access.role,
"abilities": access.get_abilities(user),
}
def test_api_team_accesses_create_anonymous():
"""Anonymous users should not be allowed to create team accesses."""
user = factories.UserFactory()
team = factories.TeamFactory()
response = APIClient().post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(user.id),
"team": str(team.id),
"role": random.choice(models.RoleChoices.choices)[0],
},
format="json",
)
assert response.status_code == 401
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
assert models.TeamAccess.objects.exists() is False
def test_api_team_accesses_create_authenticated_unrelated():
"""
Authenticated users should not be allowed to create team accesses for a team to
which they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
other_user = factories.UserFactory()
team = factories.TeamFactory()
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
},
format="json",
)
assert response.status_code == 403
assert response.json() == {
"detail": "You are not allowed to manage accesses for this team."
}
assert not models.TeamAccess.objects.filter(user=other_user).exists()
def test_api_team_accesses_create_authenticated_member():
"""Members of a team should not be allowed to create team accesses."""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "member")])
other_user = factories.UserFactory()
for role in [role[0] for role in models.RoleChoices.choices]:
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
)
assert response.status_code == 403
assert response.json() == {
"detail": "You are not allowed to manage accesses for this team."
}
assert not models.TeamAccess.objects.filter(user=other_user).exists()
def test_api_team_accesses_create_authenticated_administrator():
"""
Administrators of a team should be able to create team accesses except for the "owner" role.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
other_user = factories.UserFactory()
# It should not be allowed to create an owner access
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": "owner",
},
format="json",
)
assert response.status_code == 403
assert response.json() == {
"detail": "Only owners of a team can assign other users as owners."
}
# It should be allowed to create a lower access
role = random.choice(
[role[0] for role in models.RoleChoices.choices if role[0] != "owner"]
)
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
)
assert response.status_code == 201
assert models.TeamAccess.objects.filter(user=other_user).count() == 1
new_team_access = models.TeamAccess.objects.filter(user=other_user).get()
assert response.json() == {
"abilities": new_team_access.get_abilities(user),
"id": str(new_team_access.id),
"role": role,
"user": str(other_user.id),
}
def test_api_team_accesses_create_authenticated_owner():
"""
Owners of a team should be able to create team accesses whatever the role.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
other_user = factories.UserFactory()
role = random.choice([role[0] for role in models.RoleChoices.choices])
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
)
assert response.status_code == 201
assert models.TeamAccess.objects.filter(user=other_user).count() == 1
new_team_access = models.TeamAccess.objects.filter(user=other_user).get()
assert response.json() == {
"abilities": new_team_access.get_abilities(user),
"id": str(new_team_access.id),
"role": role,
"user": str(other_user.id),
}
def test_api_team_accesses_update_anonymous():
"""Anonymous users should not be allowed to update a team access."""
access = factories.TeamAccessFactory()
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
api_client = APIClient()
for field, value in new_values.items():
response = api_client.put(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
{**old_values, field: value},
format="json",
)
assert response.status_code == 401
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_authenticated_unrelated():
"""
Authenticated users should not be allowed to update a team access for a team to which
they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
access = factories.TeamAccessFactory()
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
for field, value in new_values.items():
response = client.put(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
{**old_values, field: value},
format="json",
)
assert response.status_code == 403
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_authenticated_member():
"""Members of a team should not be allowed to update its accesses."""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "member")])
access = factories.TeamAccessFactory(team=team)
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
for field, value in new_values.items():
response = client.put(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
{**old_values, field: value},
format="json",
)
assert response.status_code == 403
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_administrator_except_owner():
"""
A user who is an administrator in a team should be allowed to update a user
access for this team, as long as they don't try to set the role to owner.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
access = factories.TeamAccessFactory(
team=team,
role=random.choice(["administrator", "member"]),
)
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": random.choice(["administrator", "member"]),
}
for field, value in new_values.items():
new_data = {**old_values, field: value}
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data=new_data,
format="json",
)
if (
new_data["role"] == old_values["role"]
): # we are not really updating the role
assert response.status_code == 403
else:
assert response.status_code == 200
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
if field == "role":
assert updated_values == {**old_values, "role": new_values["role"]}
else:
assert updated_values == old_values
def test_api_team_accesses_update_administrator_from_owner():
"""
A user who is an administrator in a team, should not be allowed to update
the user access of an "owner" for this team.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
other_user = factories.UserFactory()
access = factories.TeamAccessFactory(team=team, user=other_user, role="owner")
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
for field, value in new_values.items():
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, field: value},
format="json",
)
assert response.status_code == 403
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_administrator_to_owner():
"""
A user who is an administrator in a team, should not be allowed to update
the user access of another user to grant team ownership.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
other_user = factories.UserFactory()
access = factories.TeamAccessFactory(
team=team,
user=other_user,
role=random.choice(["administrator", "member"]),
)
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": "owner",
}
for field, value in new_values.items():
new_data = {**old_values, field: value}
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data=new_data,
format="json",
)
# We are not allowed or not really updating the role
if field == "role" or new_data["role"] == old_values["role"]:
assert response.status_code == 403
else:
assert response.status_code == 200
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_owner_except_owner():
"""
A user who is an owner in a team should be allowed to update
a user access for this team except for existing "owner" accesses.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
factories.UserFactory()
access = factories.TeamAccessFactory(
team=team,
role=random.choice(["administrator", "member"]),
)
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
for field, value in new_values.items():
new_data = {**old_values, field: value}
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data=new_data,
format="json",
)
if (
new_data["role"] == old_values["role"]
): # we are not really updating the role
assert response.status_code == 403
else:
assert response.status_code == 200
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
if field == "role":
assert updated_values == {**old_values, "role": new_values["role"]}
else:
assert updated_values == old_values
def test_api_team_accesses_update_owner_for_owners():
"""
A user who is "owner" of a team should not be allowed to update
an existing owner access for this team.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
access = factories.TeamAccessFactory(team=team, role="owner")
old_values = serializers.TeamAccessSerializer(instance=access).data
new_values = {
"id": uuid4(),
"user_id": factories.UserFactory().id,
"role": random.choice(models.RoleChoices.choices)[0],
}
for field, value in new_values.items():
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, field: value},
content_type="application/json",
)
assert response.status_code == 403
access.refresh_from_db()
updated_values = serializers.TeamAccessSerializer(instance=access).data
assert updated_values == old_values
def test_api_team_accesses_update_owner_self():
"""
A user who is owner of a team should be allowed to update
their own user access provided there are other owners in the team.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory()
access = factories.TeamAccessFactory(team=team, user=user, role="owner")
old_values = serializers.TeamAccessSerializer(instance=access).data
new_role = random.choice(["administrator", "member"])
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, "role": new_role},
format="json",
)
assert response.status_code == 403
access.refresh_from_db()
assert access.role == "owner"
# Add another owner and it should now work
factories.TeamAccessFactory(team=team, role="owner")
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, "role": new_role},
format="json",
)
assert response.status_code == 200
access.refresh_from_db()
assert access.role == new_role
# Delete
def test_api_team_accesses_delete_anonymous():
"""Anonymous users should not be allowed to destroy a team access."""
access = factories.TeamAccessFactory()
response = APIClient().delete(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 401
assert models.TeamAccess.objects.count() == 1
def test_api_team_accesses_delete_authenticated():
"""
Authenticated users should not be allowed to delete a team access for a
team to which they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
access = factories.TeamAccessFactory()
response = client.delete(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 403
assert models.TeamAccess.objects.count() == 1
def test_api_team_accesses_delete_member():
"""
Authenticated users should not be allowed to delete a team access for a
team in which they are a simple member.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "member")])
access = factories.TeamAccessFactory(team=team)
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 403
assert models.TeamAccess.objects.count() == 2
def test_api_team_accesses_delete_administrators():
"""
Users who are administrators in a team should be allowed to delete an access
from the team provided it is not ownership.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
access = factories.TeamAccessFactory(
team=team, role=random.choice(["member", "administrator"])
)
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 204
assert models.TeamAccess.objects.count() == 1
def test_api_team_accesses_delete_owners_except_owners():
"""
Users should be able to delete the team access of another user
for a team of which they are owner provided it is not an owner access.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
access = factories.TeamAccessFactory(
team=team, role=random.choice(["member", "administrator"])
)
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 204
assert models.TeamAccess.objects.count() == 1
def test_api_team_accesses_delete_owners_for_owners():
"""
Users should not be allowed to delete the team access of another owner
even for a team in which they are direct owner.
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
access = factories.TeamAccessFactory(team=team, role="owner")
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 403
assert models.TeamAccess.objects.count() == 2
def test_api_team_accesses_delete_owners_last_owner():
"""
It should not be possible to delete the last owner access from a team
"""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
team = factories.TeamFactory()
access = factories.TeamAccessFactory(team=team, user=user, role="owner")
assert models.TeamAccess.objects.count() == 1
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
)
assert response.status_code == 403
assert models.TeamAccess.objects.count() == 1
@@ -1,424 +0,0 @@
"""
Unit tests for the Invitation model
"""
import time
import pytest
from rest_framework import status
from rest_framework.test import APIClient
from core import factories
from core.api import serializers
pytestmark = pytest.mark.django_db
def test_api_team_invitations__create__anonymous():
"""Anonymous users should not be able to create invitations."""
team = factories.TeamFactory()
invitation_values = serializers.InvitationSerializer(
factories.InvitationFactory.build()
).data
response = APIClient().post(
f"/api/v1.0/teams/{team.id}/invitations/",
invitation_values,
format="json",
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_team_invitations__create__authenticated_outsider():
"""Users outside of team should not be permitted to invite to team."""
identity = factories.IdentityFactory()
team = factories.TeamFactory()
invitation_values = serializers.InvitationSerializer(
factories.InvitationFactory.build()
).data
client = APIClient()
client.force_login(identity.user)
response = client.post(
f"/api/v1.0/teams/{team.id}/invitations/",
invitation_values,
format="json",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.parametrize(
"role",
["owner", "administrator"],
)
def test_api_team_invitations__create__privileged_members(role):
"""Owners and administrators should be able to invite new members."""
identity = factories.IdentityFactory()
team = factories.TeamFactory(users=[(identity.user, role)])
invitation_values = serializers.InvitationSerializer(
factories.InvitationFactory.build()
).data
client = APIClient()
client.force_login(identity.user)
response = client.post(
f"/api/v1.0/teams/{team.id}/invitations/",
invitation_values,
format="json",
)
assert response.status_code == status.HTTP_201_CREATED
def test_api_team_invitations__create__members():
"""
Members should not be able to invite new members.
"""
identity = factories.IdentityFactory()
team = factories.TeamFactory(users=[(identity.user, "member")])
invitation_values = serializers.InvitationSerializer(
factories.InvitationFactory.build()
).data
client = APIClient()
client.force_login(identity.user)
response = client.post(
f"/api/v1.0/teams/{team.id}/invitations/",
invitation_values,
format="json",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert response.json() == {
"detail": "You are not allowed to manage invitation for this team."
}
def test_api_team_invitations__create__issuer_and_team_automatically_added():
"""Team and issuer fields should auto-complete."""
identity = factories.IdentityFactory()
team = factories.TeamFactory(users=[(identity.user, "owner")])
# Generate a random invitation
invitation = factories.InvitationFactory.build()
invitation_data = {"email": invitation.email, "role": invitation.role}
client = APIClient()
client.force_login(identity.user)
response = client.post(
f"/api/v1.0/teams/{team.id}/invitations/",
invitation_data,
format="json",
)
assert response.status_code == status.HTTP_201_CREATED
# team and issuer automatically set
assert response.json()["team"] == str(team.id)
assert response.json()["issuer"] == str(identity.user.id)
def test_api_team_invitations__create__cannot_duplicate_invitation():
"""An email should not be invited multiple times to the same team."""
existing_invitation = factories.InvitationFactory()
team = existing_invitation.team
# Grant privileged role on the Team to the user
identity = factories.IdentityFactory()
factories.TeamAccessFactory(team=team, user=identity.user, role="administrator")
# Create a new invitation to the same team with the exact same email address
duplicated_invitation = serializers.InvitationSerializer(
factories.InvitationFactory.build(email=existing_invitation.email)
).data
client = APIClient()
client.force_login(identity.user)
response = client.post(
f"/api/v1.0/teams/{team.id}/invitations/",
duplicated_invitation,
format="json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["__all__"] == [
"Team invitation with this Email address and Team already exists."
]
def test_api_team_invitations__create__cannot_invite_existing_users():
"""
Should not be able to invite already existing users.
"""
user = factories.UserFactory()
team = factories.TeamFactory(users=[(user, "administrator")])
existing_user = factories.IdentityFactory(is_main=True)
# Build an invitation to the email of an exising identity in the db
invitation_values = serializers.InvitationSerializer(
factories.InvitationFactory.build(email=existing_user.email, team=team)
).data
client = APIClient()
client.force_login(user)
response = client.post(
f"/api/v1.0/teams/{team.id}/invitations/",
invitation_values,
format="json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["email"] == [
"This email is already associated to a registered user."
]
def test_api_team_invitations__list__anonymous_user():
"""Anonymous users should not be able to list invitations."""
team = factories.TeamFactory()
response = APIClient().get(f"/api/v1.0/teams/{team.id}/invitations/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_api_team_invitations__list__authenticated():
"""
Authenticated user should be able to list invitations
in teams they belong to, including from other issuers.
"""
identity = factories.IdentityFactory()
other_user = factories.UserFactory()
team = factories.TeamFactory(
users=[(identity.user, "administrator"), (other_user, "owner")]
)
invitation = factories.InvitationFactory(
team=team, role="administrator", issuer=identity.user
)
other_invitations = factories.InvitationFactory.create_batch(
2, team=team, role="member", issuer=other_user
)
# invitations from other teams should not be listed
other_team = factories.TeamFactory()
factories.InvitationFactory.create_batch(2, team=other_team, role="member")
client = APIClient()
client.force_login(identity.user)
response = client.get(
f"/api/v1.0/teams/{team.id}/invitations/",
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["count"] == 3
assert sorted(response.json()["results"], key=lambda x: x["created_at"]) == sorted(
[
{
"id": str(i.id),
"created_at": i.created_at.isoformat().replace("+00:00", "Z"),
"email": str(i.email),
"team": str(team.id),
"role": i.role,
"issuer": str(i.issuer.id),
"is_expired": False,
}
for i in [invitation, *other_invitations]
],
key=lambda x: x["created_at"],
)
def test_api_team_invitations__list__expired_invitations_still_listed(settings):
"""
Expired invitations are still listed.
"""
identity = factories.IdentityFactory()
other_user = factories.UserFactory()
team = factories.TeamFactory(
users=[(identity.user, "administrator"), (other_user, "owner")]
)
# override settings to accelerate validation expiration
settings.INVITATION_VALIDITY_DURATION = 1 # second
expired_invitation = factories.InvitationFactory(
team=team,
role="member",
issuer=identity.user,
)
time.sleep(1)
client = APIClient()
client.force_login(identity.user)
response = client.get(
f"/api/v1.0/teams/{team.id}/invitations/",
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["count"] == 1
assert sorted(response.json()["results"], key=lambda x: x["created_at"]) == sorted(
[
{
"id": str(expired_invitation.id),
"created_at": expired_invitation.created_at.isoformat().replace(
"+00:00", "Z"
),
"email": str(expired_invitation.email),
"team": str(team.id),
"role": expired_invitation.role,
"issuer": str(expired_invitation.issuer.id),
"is_expired": True,
},
],
key=lambda x: x["created_at"],
)
def test_api_team_invitations__retrieve__anonymous_user():
"""
Anonymous user should not be able to retrieve invitations.
"""
invitation = factories.InvitationFactory()
response = APIClient().get(
f"/api/v1.0/teams/{invitation.team.id}/invitations/{invitation.id}/",
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_api_team_invitations__retrieve__unrelated_user():
"""
Authenticated unrelated users should not be able to retrieve invitations.
"""
user = factories.IdentityFactory(user=factories.UserFactory()).user
invitation = factories.InvitationFactory()
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/teams/{invitation.team.id}/invitations/{invitation.id}/",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_api_team_invitations__retrieve__team_member():
"""
Authenticated team members should be able to retrieve invitations
whatever their role in the team.
"""
user = factories.IdentityFactory(user=factories.UserFactory()).user
invitation = factories.InvitationFactory()
factories.TeamAccessFactory(team=invitation.team, user=user, role="member")
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/teams/{invitation.team.id}/invitations/{invitation.id}/",
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"id": str(invitation.id),
"created_at": invitation.created_at.isoformat().replace("+00:00", "Z"),
"email": invitation.email,
"team": str(invitation.team.id),
"role": str(invitation.role),
"issuer": str(invitation.issuer.id),
"is_expired": False,
}
@pytest.mark.parametrize(
"method",
["put", "patch"],
)
def test_api_team_invitations__update__forbidden(method):
"""
Update of invitations is currently forbidden.
"""
user = factories.IdentityFactory(user=factories.UserFactory()).user
invitation = factories.InvitationFactory()
factories.TeamAccessFactory(team=invitation.team, user=user, role="owner")
client = APIClient()
client.force_login(user)
response = None
if method == "put":
response = client.put(
f"/api/v1.0/teams/{invitation.team.id}/invitations/{invitation.id}/",
)
if method == "patch":
response = client.patch(
f"/api/v1.0/teams/{invitation.team.id}/invitations/{invitation.id}/",
)
assert response is not None
assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
assert response.json()["detail"] == f'Method "{method.upper()}" not allowed.'
def test_api_team_invitations__delete__anonymous():
"""Anonymous user should not be able to delete invitations."""
invitation = factories.InvitationFactory()
response = APIClient().delete(
f"/api/v1.0/teams/{invitation.team.id}/invitations/{invitation.id}/",
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_api_team_invitations__delete__authenticated_outsider():
"""Members outside of team should not cancel invitations."""
identity = factories.IdentityFactory()
team = factories.TeamFactory()
invitation = factories.InvitationFactory(team=team)
client = APIClient()
client.force_login(identity.user)
response = client.delete(
f"/api/v1.0/teams/{team.id}/invitations/{invitation.id}/",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
@pytest.mark.parametrize("role", ["owner", "administrator"])
def test_api_team_invitations__delete__privileged_members(role):
"""Privileged member should be able to cancel invitation."""
identity = factories.IdentityFactory()
team = factories.TeamFactory(users=[(identity.user, role)])
invitation = factories.InvitationFactory(team=team)
client = APIClient()
client.force_login(identity.user)
response = client.delete(
f"/api/v1.0/teams/{team.id}/invitations/{invitation.id}/",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
def test_api_team_invitations__delete__members():
"""Member should not be able to cancel invitation."""
identity = factories.IdentityFactory()
team = factories.TeamFactory(users=[(identity.user, "member")])
invitation = factories.InvitationFactory(team=team)
client = APIClient()
client.force_login(identity.user)
response = client.delete(
f"/api/v1.0/teams/{team.id}/invitations/{invitation.id}/",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert (
response.json()["detail"]
== "You do not have permission to perform this action."
)
+32 -304
View File
@@ -1,7 +1,6 @@
"""
Test users API endpoints in the People core app.
"""
from unittest import mock
import pytest
@@ -53,22 +52,16 @@ def test_api_users_authenticated_list_by_email():
Authenticated users should be able to search users with a case-insensitive and
partial query on the email.
"""
user = factories.UserFactory(admin_email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.admin_email, name="john doe")
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
client = APIClient()
client.force_login(user)
dave = factories.IdentityFactory(
email="david.bowman@work.com", name=None, is_main=True
)
nicole = factories.IdentityFactory(
email="nicole_foole@work.com", name=None, is_main=True
)
frank = factories.IdentityFactory(
email="frank_poole@work.com", name=None, is_main=True
)
factories.IdentityFactory(email="heywood_floyd@work.com", name=None, is_main=True)
dave = factories.IdentityFactory(email="david.bowman@work.com")
nicole = factories.IdentityFactory(email="nicole_foole@work.com")
frank = factories.IdentityFactory(email="frank_poole@work.com")
factories.IdentityFactory(email="heywood_floyd@work.com")
# Full query should work
response = client.get(
@@ -97,200 +90,24 @@ def test_api_users_authenticated_list_by_email():
# Even with a low similarity threshold, query should match expected emails
response = client.get("/api/v1.0/users/?q=ool")
assert response.status_code == HTTP_200_OK
assert response.json()["results"] == [
{
"id": str(nicole.user.id),
"email": nicole.email,
"name": nicole.name,
"is_device": nicole.user.is_device,
"is_staff": nicole.user.is_staff,
"language": nicole.user.language,
"timezone": str(nicole.user.timezone),
},
{
"id": str(frank.user.id),
"email": frank.email,
"name": frank.name,
"is_device": frank.user.is_device,
"is_staff": frank.user.is_staff,
"language": frank.user.language,
"timezone": str(frank.user.timezone),
},
]
def test_api_users_authenticated_list_by_name():
"""
Authenticated users should be able to search users with a case-insensitive and
partial query on the name.
"""
user = factories.UserFactory(admin_email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.admin_email, name="john doe")
client = APIClient()
client.force_login(user)
dave = factories.IdentityFactory(name="dave bowman", email=None, is_main=True)
nicole = factories.IdentityFactory(name="nicole foole", email=None, is_main=True)
frank = factories.IdentityFactory(name="frank poole", email=None, is_main=True)
factories.IdentityFactory(name="heywood floyd", email=None, is_main=True)
# Full query should work
response = client.get(
"/api/v1.0/users/?q=david.bowman@work.com",
)
assert response.status_code == HTTP_200_OK
user_ids = [user["id"] for user in response.json()["results"]]
assert user_ids[0] == str(dave.user.id)
# Partial query should work
response = client.get("/api/v1.0/users/?q=fran")
assert response.status_code == HTTP_200_OK
user_ids = [user["id"] for user in response.json()["results"]]
assert user_ids[0] == str(frank.user.id)
# Result that matches a trigram twice ranks better than result that matches once
response = client.get("/api/v1.0/users/?q=ole")
assert response.status_code == HTTP_200_OK
user_ids = [user["id"] for user in response.json()["results"]]
# "Nicole Foole" matches twice on "ole"
assert user_ids == [str(nicole.user.id), str(frank.user.id)]
# Even with a low similarity threshold, query should match expected user
response = client.get("/api/v1.0/users/?q=ool")
assert response.status_code == HTTP_200_OK
assert response.json()["results"] == [
{
"id": str(nicole.user.id),
"email": nicole.email,
"name": nicole.name,
"is_device": nicole.user.is_device,
"is_staff": nicole.user.is_staff,
"language": nicole.user.language,
"timezone": str(nicole.user.timezone),
},
{
"id": str(frank.user.id),
"email": frank.email,
"name": frank.name,
"is_device": frank.user.is_device,
"is_staff": frank.user.is_staff,
"language": frank.user.language,
"timezone": str(frank.user.timezone),
},
]
def test_api_users_authenticated_list_by_name_and_email():
"""
Authenticated users should be able to search users with a case-insensitive and
partial query on the name and email.
"""
user = factories.UserFactory(admin_email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.admin_email, name="john doe")
client = APIClient()
client.force_login(user)
nicole = factories.IdentityFactory(
email="nicole_foole@work.com", name="nicole foole", is_main=True
)
frank = factories.IdentityFactory(
email="oleg_poole@work.com", name=None, is_main=True
)
david = factories.IdentityFactory(email=None, name="david role", is_main=True)
# Result that matches a trigram in name and email ranks better than result that matches once
response = client.get("/api/v1.0/users/?q=ole")
assert response.status_code == HTTP_200_OK
user_ids = [user["id"] for user in response.json()["results"]]
# "Nicole Foole" matches twice on "ole" in her name and twice on "ole" in her email
# "Oleg poole" matches twice on "ole" in her email
# "David role" matches once on "ole" in his name
assert user_ids == [str(nicole.user.id), str(frank.user.id), str(david.user.id)]
def test_api_users_authenticated_list_exclude_users_already_in_team(
django_assert_num_queries,
):
"""
Authenticated users should be able to search users
but the result should exclude all users already in the given team.
"""
user = factories.UserFactory(admin_email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email, name="john doe")
client = APIClient()
client.force_login(user)
dave = factories.IdentityFactory(name="dave bowman", email=None, is_main=True)
nicole = factories.IdentityFactory(name="nicole foole", email=None, is_main=True)
frank = factories.IdentityFactory(name="frank poole", email=None, is_main=True)
mary = factories.IdentityFactory(name="mary poole", email=None, is_main=True)
factories.IdentityFactory(name="heywood floyd", email=None, is_main=True)
factories.IdentityFactory(name="Andrei Smyslov", email=None, is_main=True)
factories.TeamFactory.create_batch(10)
# Add Dave and Frank in the same team
team = factories.TeamFactory(
users=[
(dave.user, models.RoleChoices.MEMBER),
(frank.user, models.RoleChoices.MEMBER),
]
)
factories.TeamFactory(users=[(nicole.user, models.RoleChoices.MEMBER)])
# Search user to add him/her to a team, we give a team id to the request
# to exclude all users already in the team
# We can't find David Bowman because he is already a member of the given team
# 2 queries are needed here:
# - user authenticated
# - search user query
with django_assert_num_queries(2):
response = client.get(
f"/api/v1.0/users/?q=bowman&team_id={team.id}",
)
assert response.status_code == HTTP_200_OK
assert response.json()["results"] == []
# We can only find Nicole and Mary because Frank is already a member of the given team
# 4 queries are needed here:
# - user authenticated
# - search user query
# - User
# - Identity
with django_assert_num_queries(4):
response = client.get(
f"/api/v1.0/users/?q=ool&team_id={team.id}",
)
assert response.status_code == HTTP_200_OK
user_ids = sorted([user["id"] for user in response.json()["results"]])
assert user_ids == sorted([str(mary.user.id), str(nicole.user.id)])
def test_api_users_authenticated_list_multiple_identities_single_user():
"""
User with multiple identities should appear only once in results.
"""
user = factories.UserFactory(admin_email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.admin_email, name="eva karl")
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
client = APIClient()
client.force_login(user)
dave = factories.UserFactory()
factories.IdentityFactory(
user=dave, email="dave.bowman@work.com", name="dave bowman"
)
factories.IdentityFactory(user=dave, email="dave.bowman@fun.fr", name="dave bowman")
factories.IdentityFactory(user=dave, email="david.bowman@work.com")
factories.IdentityFactory(user=dave, email="david.bowman@fun.fr")
# Full query should work
response = client.get(
@@ -308,25 +125,19 @@ def test_api_users_authenticated_list_multiple_identities_multiple_users():
User with multiple identities should be ranked
on their best matching identity.
"""
user = factories.UserFactory(admin_email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.admin_email, name="john doe")
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
client = APIClient()
client.force_login(user)
dave = factories.UserFactory()
dave_identity = factories.IdentityFactory(
user=dave, email="dave.bowman@work.com", is_main=True, name="dave bowman"
)
factories.IdentityFactory(user=dave, email="babibou@ehehe.com", name="babihou")
davina_identity = factories.IdentityFactory(
user=factories.UserFactory(), email="davina.bowan@work.com", name="davina"
)
prue_identity = factories.IdentityFactory(
user=factories.UserFactory(),
email="prudence.crandall@work.com",
name="prudence",
)
davina = factories.UserFactory()
prudence = factories.UserFactory()
factories.IdentityFactory(user=dave, email="david.bowman@work.com")
factories.IdentityFactory(user=dave, email="babibou@ehehe.com")
factories.IdentityFactory(user=davina, email="davina.bowan@work.com")
factories.IdentityFactory(user=prudence, email="prudence.crandall@work.com")
# Full query should work
response = client.get(
@@ -335,48 +146,19 @@ def test_api_users_authenticated_list_multiple_identities_multiple_users():
assert response.status_code == HTTP_200_OK
assert response.json()["count"] == 3
assert response.json()["results"] == [
{
"id": str(dave.id),
"email": dave_identity.email,
"name": dave_identity.name,
"is_device": dave_identity.user.is_device,
"is_staff": dave_identity.user.is_staff,
"language": dave_identity.user.language,
"timezone": str(dave_identity.user.timezone),
},
{
"id": str(davina_identity.user.id),
"email": davina_identity.email,
"name": davina_identity.name,
"is_device": davina_identity.user.is_device,
"is_staff": davina_identity.user.is_staff,
"language": davina_identity.user.language,
"timezone": str(davina_identity.user.timezone),
},
{
"id": str(prue_identity.user.id),
"email": prue_identity.email,
"name": prue_identity.name,
"is_device": prue_identity.user.is_device,
"is_staff": prue_identity.user.is_staff,
"language": prue_identity.user.language,
"timezone": str(prue_identity.user.timezone),
},
]
user_ids = [user["id"] for user in response.json()["results"]]
assert user_ids == [str(dave.id), str(davina.id), str(prudence.id)]
def test_api_users_authenticated_list_uppercase_content():
"""Upper case content should be found by lower case query."""
user = factories.UserFactory(admin_email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.admin_email, name="eva karl")
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
client = APIClient()
client.force_login(user)
dave = factories.IdentityFactory(
email="DAVID.BOWMAN@INTENSEWORK.COM", name="DAVID BOWMAN"
)
dave = factories.IdentityFactory(email="DAVID.BOWMAN@INTENSEWORK.COM")
# Unaccented full address
response = client.get(
@@ -399,13 +181,13 @@ def test_api_users_authenticated_list_uppercase_content():
def test_api_users_list_authenticated_capital_query():
"""Upper case query should find lower case content."""
user = factories.UserFactory(admin_email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.admin_email, name="eva karl")
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
client = APIClient()
client.force_login(user)
dave = factories.IdentityFactory(email="david.bowman@work.com", name="david bowman")
dave = factories.IdentityFactory(email="david.bowman@work.com")
# Full uppercase query
response = client.get(
@@ -428,15 +210,13 @@ def test_api_users_list_authenticated_capital_query():
def test_api_contacts_list_authenticated_accented_query():
"""Accented content should be found by unaccented query."""
user = factories.UserFactory(admin_email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.admin_email, name="john doe")
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
client = APIClient()
client.force_login(user)
helene = factories.IdentityFactory(
email="helene.bowman@work.com", name="helene bowman"
)
helene = factories.IdentityFactory(email="helene.bowman@work.com")
# Accented full query
response = client.get(
@@ -498,58 +278,6 @@ def test_api_users_list_pagination(
assert len(content["results"]) == 2
@pytest.mark.parametrize("page_size", [1, 10, 99, 100])
def test_api_users_list_pagination_page_size(
page_size,
):
"""Page's size on pagination should work as expected."""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
for i in range(page_size):
factories.UserFactory.create(admin_email=f"user-{i}@people.com")
response = client.get(
f"/api/v1.0/users/?page_size={page_size}",
)
assert response.status_code == HTTP_200_OK
content = response.json()
assert content["count"] == page_size + 1
assert len(content["results"]) == page_size
@pytest.mark.parametrize("page_size", [101, 200])
def test_api_users_list_pagination_wrong_page_size(
page_size,
):
"""Page's size on pagination should be limited to "max_page_size"."""
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
for i in range(page_size):
factories.UserFactory.create(admin_email=f"user-{i}@people.com")
response = client.get(
f"/api/v1.0/users/?page_size={page_size}",
)
assert response.status_code == HTTP_200_OK
content = response.json()
assert content["count"] == page_size + 1
# Length should not exceed "max_page_size" default value
assert len(content["results"]) == 100
def test_api_users_retrieve_me_anonymous():
"""Anonymous users should not be allowed to list users."""
factories.UserFactory.create_batch(2)
@@ -563,7 +291,7 @@ def test_api_users_retrieve_me_anonymous():
def test_api_users_retrieve_me_authenticated():
"""Authenticated users should be able to retrieve their own user via the "/users/me" path."""
identity = factories.IdentityFactory(is_main=True)
identity = factories.IdentityFactory()
user = identity.user
client = APIClient()
@@ -582,12 +310,12 @@ def test_api_users_retrieve_me_authenticated():
assert response.status_code == HTTP_200_OK
assert response.json() == {
"id": str(user.id),
"name": str(identity.name),
"email": str(identity.email),
"email": str(user.email),
"language": user.language,
"timezone": str(user.timezone),
"is_device": False,
"is_staff": False,
"data": user.profile_contact.data,
}
@@ -1,11 +1,12 @@
"""Unit tests for the Authentication Backends."""
from django.core.exceptions import SuspiciousOperation
"""Unit tests for the `get_or_create_user` function."""
from unittest import mock
import pytest
from rest_framework_simplejwt.exceptions import InvalidToken
from rest_framework_simplejwt.tokens import AccessToken
from core import models
from core.authentication.backends import OIDCAuthenticationBackend
from core import factories, models
from core.authentication import OIDCAuthenticationBackend
from core.factories import IdentityFactory
pytestmark = pytest.mark.django_db
@@ -21,7 +22,7 @@ def test_authentication_getter_existing_user_no_email(
klass = OIDCAuthenticationBackend()
# Create a user and its identity
identity = IdentityFactory(name=None)
identity = IdentityFactory()
# Create multiple identities for a user
for _ in range(5):
@@ -34,7 +35,7 @@ def test_authentication_getter_existing_user_no_email(
with django_assert_num_queries(1):
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
access_token=AccessToken(), id_token=None, payload=None
)
identity.refresh_from_db()
@@ -46,86 +47,49 @@ def test_authentication_getter_existing_user_with_email(
):
"""
When the user's info contains an email and targets an existing user,
it should update the email on the identity but not on the user.
"""
klass = OIDCAuthenticationBackend()
identity = IdentityFactory(name="John Doe")
identity = IdentityFactory()
# Create multiple identities for a user
for _ in range(5):
IdentityFactory(user=identity.user)
user_email = identity.user.email
assert models.User.objects.count() == 1
def get_userinfo_mocked(*args):
return {
"sub": identity.sub,
"email": identity.email,
"first_name": "John",
"last_name": "Doe",
}
return {"sub": identity.sub, "email": identity.email}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
# Only 1 query because email and names have not changed
# Only 1 query if the email has not changed
with django_assert_num_queries(1):
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
access_token=AccessToken(), id_token=None, payload=None
)
assert models.User.objects.get() == user
@pytest.mark.parametrize(
"first_name, last_name, email",
[
("Jack", "Doe", "john.doe@example.com"),
("John", "Duy", "john.doe@example.com"),
("John", "Doe", "jack.duy@example.com"),
("Jack", "Duy", "jack.duy@example.com"),
],
)
def test_authentication_getter_existing_user_change_fields(
first_name, last_name, email, django_assert_num_queries, monkeypatch
):
"""
It should update the email or name fields on the identity when they change.
The email on the user should not be changed.
"""
klass = OIDCAuthenticationBackend()
identity = IdentityFactory(name="John Doe", email="john.doe@example.com")
user_email = identity.user.admin_email
# Create multiple identities for a user
for _ in range(5):
IdentityFactory(user=identity.user)
assert models.User.objects.count() == 1
new_email = "test@fooo.com"
def get_userinfo_mocked(*args):
return {
"sub": identity.sub,
"email": email,
"first_name": first_name,
"last_name": last_name,
}
return {"sub": identity.sub, "email": new_email}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
# One and only one additional update query when a field has changed
# Additional update query if the email has changed
with django_assert_num_queries(2):
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
access_token=AccessToken(), id_token=None, payload=None
)
identity.refresh_from_db()
assert identity.email == email
assert identity.name == f"{first_name:s} {last_name:s}"
assert identity.email == new_email
assert models.User.objects.count() == 1
assert user == identity.user
assert user.admin_email == user_email
assert user.email == user_email
def test_authentication_getter_new_user_no_email(monkeypatch):
@@ -141,14 +105,14 @@ def test_authentication_getter_new_user_no_email(monkeypatch):
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
access_token=AccessToken(), id_token=None, payload=None
)
identity = user.identities.get()
assert identity.sub == "123"
assert identity.email is None
assert user.admin_email is None
assert user.email is None
assert user.password == "!"
assert models.User.objects.count() == 1
@@ -156,28 +120,26 @@ def test_authentication_getter_new_user_no_email(monkeypatch):
def test_authentication_getter_new_user_with_email(monkeypatch):
"""
If no user matches the user's info sub, a user should be created.
User's email and name should be set on the identity.
The "email" field on the User model should not be set as it is reserved for staff users.
User's info contains an email, created user's email should be set.
"""
klass = OIDCAuthenticationBackend()
email = "people@example.com"
def get_userinfo_mocked(*args):
return {"sub": "123", "email": email, "first_name": "John", "last_name": "Doe"}
return {"sub": "123", "email": email}
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
user = klass.get_or_create_user(
access_token="test-token", id_token=None, payload=None
access_token=AccessToken(), id_token=None, payload=None
)
identity = user.identities.get()
assert identity.sub == "123"
assert identity.email == email
assert identity.name == "John Doe"
assert user.admin_email is None
assert user.email == email
assert models.User.objects.count() == 1
@@ -192,13 +154,11 @@ def test_models_oidc_user_getter_invalid_token(django_assert_num_queries, monkey
monkeypatch.setattr(OIDCAuthenticationBackend, "get_userinfo", get_userinfo_mocked)
with (
django_assert_num_queries(0),
pytest.raises(
SuspiciousOperation,
match="User info contained no recognizable user identification",
),
with django_assert_num_queries(0), pytest.raises(
InvalidToken, match="User info contained no recognizable user identification"
):
klass.get_or_create_user(access_token="test-token", id_token=None, payload=None)
user = klass.get_or_create_user(
access_token=AccessToken(), id_token=None, payload=None
)
assert models.User.objects.exists() is False
@@ -1,7 +1,6 @@
"""
Unit tests for the Contact model
"""
from django.core.exceptions import ValidationError
import pytest
@@ -1,7 +1,6 @@
"""
Unit tests for the Identity model
"""
from django.core.exceptions import ValidationError
import pytest
@@ -71,19 +70,19 @@ def test_models_identities_is_main_switch():
def test_models_identities_email_not_required():
"""The 'email' field can be blank."""
"""The "email" field can be blank."""
user = factories.UserFactory()
models.Identity.objects.create(user=user, sub="123", email=None)
def test_models_identities_user_required():
"""The 'user' field is required."""
"""The "user" field is required."""
with pytest.raises(models.User.DoesNotExist, match="Identity has no user."):
models.Identity.objects.create(user=None, email="david@example.com")
def test_models_identities_email_unique_same_user():
"""The 'email' field should be unique for a given user."""
"""The "email" field should be unique for a given user."""
email = factories.IdentityFactory()
with pytest.raises(
@@ -94,13 +93,13 @@ def test_models_identities_email_unique_same_user():
def test_models_identities_email_unique_different_users():
"""The 'email' field should not be unique among users."""
"""The "email" field should not be unique among users."""
email = factories.IdentityFactory()
factories.IdentityFactory(email=email.email)
def test_models_identities_email_normalization():
"""The 'email' field should be automatically normalized upon saving."""
"""The email field should be automatically normalized upon saving."""
email = factories.IdentityFactory()
email.email = "Thomas.Jefferson@Example.com"
email.save()
@@ -121,21 +120,21 @@ def test_models_identities_ordering():
def test_models_identities_sub_null():
"""The 'sub' field should not be null."""
"""The "sub" field should not be null."""
user = factories.UserFactory()
with pytest.raises(ValidationError, match="This field cannot be null."):
models.Identity.objects.create(user=user, sub=None)
def test_models_identities_sub_blank():
"""The 'sub' field should not be blank."""
"""The "sub" field should not be blank."""
user = factories.UserFactory()
with pytest.raises(ValidationError, match="This field cannot be blank."):
models.Identity.objects.create(user=user, email="david@example.com", sub="")
def test_models_identities_sub_unique():
"""The 'sub' field should be unique."""
"""The "sub" field should be unique."""
user = factories.UserFactory()
identity = factories.IdentityFactory()
with pytest.raises(ValidationError, match="Identity with this Sub already exists."):
@@ -143,7 +142,7 @@ def test_models_identities_sub_unique():
def test_models_identities_sub_max_length():
"""The 'sub' field should be 255 characters maximum."""
"""The subfield should be 255 characters maximum."""
factories.IdentityFactory(sub="a" * 255)
with pytest.raises(ValidationError) as excinfo:
factories.IdentityFactory(sub="a" * 256)
@@ -155,13 +154,13 @@ def test_models_identities_sub_max_length():
def test_models_identities_sub_special_characters():
"""The 'sub' field should accept periods, dashes, +, @ and underscores."""
"""The subfield should accept periods, dashes, +, @ and underscores."""
identity = factories.IdentityFactory(sub="dave.bowman-1+2@hal_9000")
assert identity.sub == "dave.bowman-1+2@hal_9000"
def test_models_identities_sub_spaces():
"""The 'sub' field should not accept spaces."""
"""The subfield should not accept spaces."""
with pytest.raises(ValidationError) as excinfo:
factories.IdentityFactory(sub="a b")
@@ -172,12 +171,12 @@ def test_models_identities_sub_spaces():
def test_models_identities_sub_upper_case():
"""The 'sub' field should accept upper case characters."""
"""The subfield should accept upper case characters."""
identity = factories.IdentityFactory(sub="John")
assert identity.sub == "John"
def test_models_identities_sub_ascii():
"""The 'sub' field should accept non ASCII letters."""
"""The subfield should accept non ASCII letters."""
identity = factories.IdentityFactory(sub="rené")
assert identity.sub == "rené"
@@ -1,303 +0,0 @@
"""
Unit tests for the Invitation model
"""
import smtplib
import time
import uuid
from logging import Logger
from unittest import mock
from django.contrib.auth.models import AnonymousUser
from django.core import exceptions, mail
import pytest
from faker import Faker
from freezegun import freeze_time
from core import factories, models
pytestmark = pytest.mark.django_db
fake = Faker()
def test_models_invitations_readonly_after_create():
"""Existing invitations should be readonly."""
invitation = factories.InvitationFactory()
with pytest.raises(exceptions.PermissionDenied):
invitation.save()
def test_models_invitations_email_no_empty_mail():
"""The "email" field should not be empty."""
with pytest.raises(exceptions.ValidationError, match="This field cannot be blank"):
factories.InvitationFactory(email="")
def test_models_invitations_email_no_null_mail():
"""The "email" field is required."""
with pytest.raises(exceptions.ValidationError, match="This field cannot be null"):
factories.InvitationFactory(email=None)
def test_models_invitations_team_required():
"""The "team" field is required."""
with pytest.raises(exceptions.ValidationError, match="This field cannot be null"):
factories.InvitationFactory(team=None)
def test_models_invitations_team_should_be_team_instance():
"""The "team" field should be a team instance."""
with pytest.raises(ValueError, match='Invitation.team" must be a "Team" instance'):
factories.InvitationFactory(team="ee")
def test_models_invitations_role_required():
"""The "role" field is required."""
with pytest.raises(exceptions.ValidationError, match="This field cannot be blank"):
factories.InvitationFactory(role="")
def test_models_invitations_role_among_choices():
"""The "role" field should be a valid choice."""
with pytest.raises(
exceptions.ValidationError, match="Value 'boss' is not a valid choice"
):
factories.InvitationFactory(role="boss")
def test_models_invitations__is_expired(settings):
"""
The 'is_expired' property should return False until validity duration
is exceeded and True afterwards.
"""
expired_invitation = factories.InvitationFactory()
assert expired_invitation.is_expired is False
settings.INVITATION_VALIDITY_DURATION = 1
time.sleep(1)
assert expired_invitation.is_expired is True
def test_models_invitation__new_user__convert_invitations_to_accesses():
"""
Upon creating a new identity, invitations linked to that email
should be converted to accesses and then deleted.
"""
# Two invitations to the same mail but to different teams
invitation_to_team1 = factories.InvitationFactory()
invitation_to_team2 = factories.InvitationFactory(email=invitation_to_team1.email)
other_invitation = factories.InvitationFactory(
team=invitation_to_team2.team
) # another person invited to team2
new_identity = factories.IdentityFactory(
is_main=True, email=invitation_to_team1.email
)
# The invitation regarding
assert models.TeamAccess.objects.filter(
team=invitation_to_team1.team, user=new_identity.user
).exists()
assert models.TeamAccess.objects.filter(
team=invitation_to_team2.team, user=new_identity.user
).exists()
assert not models.Invitation.objects.filter(
team=invitation_to_team1.team, email=invitation_to_team1.email
).exists() # invitation "consumed"
assert not models.Invitation.objects.filter(
team=invitation_to_team2.team, email=invitation_to_team2.email
).exists() # invitation "consumed"
assert models.Invitation.objects.filter(
team=invitation_to_team2.team, email=other_invitation.email
).exists() # the other invitation remains
def test_models_invitation__new_user__filter_expired_invitations():
"""
Upon creating a new identity, valid invitations should be converted into accesses
and expired invitations should remain unchanged.
"""
with freeze_time("2020-01-01"):
expired_invitation = factories.InvitationFactory()
user_email = expired_invitation.email
valid_invitation = factories.InvitationFactory(email=user_email)
new_identity = factories.IdentityFactory(is_main=True, email=user_email)
# valid invitation should have granted access to the related team
assert models.TeamAccess.objects.filter(
team=valid_invitation.team, user=new_identity.user
).exists()
assert not models.Invitation.objects.filter(
team=valid_invitation.team, email=user_email
).exists()
# expired invitation should not have been consumed
assert not models.TeamAccess.objects.filter(
team=expired_invitation.team, user=new_identity.user
).exists()
assert models.Invitation.objects.filter(
team=expired_invitation.team, email=user_email
).exists()
@pytest.mark.parametrize("num_invitations, num_queries", [(0, 8), (1, 11), (20, 11)])
def test_models_invitation__new_user__user_creation_constant_num_queries(
django_assert_num_queries, num_invitations, num_queries
):
"""
The number of queries executed during user creation should not be proportional
to the number of invitations being processed.
"""
user_email = fake.email()
if num_invitations != 0:
for _ in range(0, num_invitations):
factories.InvitationFactory(email=user_email, team=factories.TeamFactory())
user = factories.UserFactory()
# with no invitation, we skip an "if", resulting in 8 requests
# otherwise, we should have 11 queries with any number of invitations
with django_assert_num_queries(num_queries):
models.Identity.objects.create(
is_main=True,
email=user_email,
user=user,
name="Prudence C.",
sub=uuid.uuid4(),
)
# get_abilities
def test_models_team_invitations_get_abilities_anonymous():
"""Check abilities returned for an anonymous user."""
access = factories.InvitationFactory()
abilities = access.get_abilities(AnonymousUser())
assert abilities == {
"delete": False,
"get": False,
"patch": False,
"put": False,
}
def test_models_team_invitations_get_abilities_authenticated():
"""Check abilities returned for an authenticated user."""
access = factories.InvitationFactory()
user = factories.UserFactory()
abilities = access.get_abilities(user)
assert abilities == {
"delete": False,
"get": False,
"patch": False,
"put": False,
}
@pytest.mark.parametrize("role", ["administrator", "owner"])
def test_models_team_invitations_get_abilities_privileged_member(role):
"""Check abilities for a team member with a privileged role."""
pivileged_access = factories.TeamAccessFactory(role=role)
team = pivileged_access.team
factories.TeamAccessFactory(team=team) # another one
invitation = factories.InvitationFactory(team=team)
abilities = invitation.get_abilities(pivileged_access.user)
assert abilities == {
"delete": True,
"get": True,
"patch": False,
"put": False,
}
def test_models_team_invitations_get_abilities_member():
"""Check abilities for a team member with 'member' role."""
member_access = factories.TeamAccessFactory(role="member")
team = member_access.team
factories.TeamAccessFactory(team=team) # another one
invitation = factories.InvitationFactory(team=team)
abilities = invitation.get_abilities(member_access.user)
assert abilities == {
"delete": False,
"get": True,
"patch": False,
"put": False,
}
def test_models_team_invitations_email():
"""Check email invitation during invitation creation."""
member_access = factories.TeamAccessFactory(role="member")
team = member_access.team
# pylint: disable-next=no-member
assert len(mail.outbox) == 0
factories.TeamAccessFactory(team=team)
invitation = factories.InvitationFactory(team=team, email="john@people.com")
# pylint: disable-next=no-member
assert len(mail.outbox) == 1
# pylint: disable-next=no-member
email = mail.outbox[0]
assert email.to == [invitation.email]
assert email.subject == "Invitation to join Desk!"
email_content = " ".join(email.body.split())
assert "Invitation to join Desk!" in email_content
assert "[//example.com]" in email_content
@mock.patch(
"django.core.mail.send_mail",
side_effect=smtplib.SMTPException("Error SMTPException"),
)
@mock.patch.object(Logger, "error")
def test_models_team_invitations_email_failed(mock_logger, _mock_send_mail):
"""Check invitation behavior when an SMTP error occurs during invitation creation."""
member_access = factories.TeamAccessFactory(role="member")
team = member_access.team
# pylint: disable-next=no-member
assert len(mail.outbox) == 0
factories.TeamAccessFactory(team=team)
# No error should be raised
invitation = factories.InvitationFactory(team=team, email="john@people.com")
# No email has been sent
# pylint: disable-next=no-member
assert len(mail.outbox) == 0
# Logger should be called
mock_logger.assert_called_once()
(
_,
email,
exception,
) = mock_logger.call_args.args
assert email == invitation.email
assert isinstance(exception, smtplib.SMTPException)
@@ -1,17 +1,12 @@
"""
Unit tests for the TeamAccess model
"""
import json
import re
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import ValidationError
import pytest
import responses
from core import factories, models
from core import factories
pytestmark = pytest.mark.django_db
@@ -43,94 +38,6 @@ def test_models_team_accesses_unique():
factories.TeamAccessFactory(user=access.user, team=access.team)
def test_models_team_accesses_create_webhook():
"""
When the team has a webhook, creating a team access should fire a call.
"""
user = factories.UserFactory()
team = factories.TeamFactory()
webhook = factories.TeamWebhookFactory(team=team)
with responses.RequestsMock() as rsps:
# Ensure successful response by scim provider using "responses":
rsp = rsps.add(
rsps.PATCH,
re.compile(r".*/Groups/.*"),
body="{}",
status=200,
content_type="application/json",
)
models.TeamAccess.objects.create(user=user, team=team)
assert rsp.call_count == 1
assert rsps.calls[0].request.url == webhook.url
# Payload sent to scim provider
payload = json.loads(rsps.calls[0].request.body)
assert payload == {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "add",
"path": "members",
"value": [
{
"value": str(user.id),
"email": None,
"type": "User",
}
],
}
],
}
def test_models_team_accesses_delete_webhook():
"""
When the team has a webhook, deleting a team access should fire a call.
"""
team = factories.TeamFactory()
webhook = factories.TeamWebhookFactory(team=team)
access = factories.TeamAccessFactory(team=team)
with responses.RequestsMock() as rsps:
# Ensure successful response by scim provider using "responses":
rsp = rsps.add(
rsps.PATCH,
re.compile(r".*/Groups/.*"),
body="{}",
status=200,
content_type="application/json",
)
access.delete()
assert rsp.call_count == 1
assert rsps.calls[0].request.url == webhook.url
# Payload sent to scim provider
payload = json.loads(rsps.calls[0].request.body)
assert payload == {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "remove",
"path": "members",
"value": [
{
"value": str(access.user.id),
"email": None,
"type": "User",
}
],
}
],
}
assert models.TeamAccess.objects.exists() is False
# get_abilities
+2 -3
View File
@@ -1,7 +1,6 @@
"""
Unit tests for the Team model
"""
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import ValidationError
@@ -62,7 +61,7 @@ def test_models_teams_get_abilities_anonymous():
abilities = team.get_abilities(AnonymousUser())
assert abilities == {
"delete": False,
"get": False,
"get": True,
"patch": False,
"put": False,
"manage_accesses": False,
@@ -75,7 +74,7 @@ def test_models_teams_get_abilities_authenticated():
abilities = team.get_abilities(factories.UserFactory())
assert abilities == {
"delete": False,
"get": False,
"get": True,
"patch": False,
"put": False,
"manage_accesses": False,
+9 -25
View File
@@ -1,7 +1,6 @@
"""
Unit tests for the User model
"""
from unittest import mock
from django.core.exceptions import ValidationError
@@ -31,20 +30,20 @@ def test_models_users_id_unique():
def test_models_users_email_unique():
"""The "admin_email" field should be unique except for the null value."""
"""The "email" field should be unique except for the null value."""
user = factories.UserFactory()
with pytest.raises(
ValidationError, match="User with this Admin email address already exists."
ValidationError, match="User with this Email address already exists."
):
models.User.objects.create(admin_email=user.admin_email, password="password")
factories.UserFactory(email=user.email)
def test_models_users_email_several_null():
"""Several users with a null value for the "email" field can co-exist."""
factories.UserFactory(admin_email=None)
models.User.objects.create(admin_email=None, password="foo.")
factories.UserFactory(email=None)
factories.UserFactory(email=None)
assert models.User.objects.filter(admin_email__isnull=True).count() == 2
assert models.User.objects.filter(email__isnull=True).count() == 2
def test_models_users_profile_not_owned():
@@ -91,26 +90,11 @@ def test_models_users_send_mail_main_existing():
)
def test_models_users_send_mail_main_admin():
"""
The 'email_user' method should send mail to the user's admin email address if the
user has no related identities.
"""
user = factories.UserFactory()
with mock.patch("django.core.mail.send_mail") as mock_send:
user.email_user("my subject", "my message")
mock_send.assert_called_once_with(
"my subject", "my message", None, [user.admin_email]
)
def test_models_users_send_mail_main_missing():
"""The 'email_user' method should fail if the user has no email address."""
user = factories.UserFactory(admin_email=None)
user = factories.UserFactory()
with pytest.raises(ValueError) as excinfo:
with pytest.raises(models.Identity.DoesNotExist) as excinfo:
user.email_user("my subject", "my message")
assert str(excinfo.value) == "You must first set an email for the user."
assert str(excinfo.value) == "Identity matching query does not exist."
-1
View File
@@ -1,7 +1,6 @@
"""
Test Throttle in People's app.
"""
import pytest
from rest_framework.test import APIClient
@@ -1,341 +0,0 @@
"""Test Team synchronization webhooks."""
import json
import random
import re
from logging import Logger
from unittest import mock
import pytest
import responses
from core import factories
from core.utils.webhooks import scim_synchronizer
pytestmark = pytest.mark.django_db
def test_utils_webhooks_add_user_to_group_no_webhooks():
"""If no webhook is declared on the team, the function should not make any request."""
access = factories.TeamAccessFactory()
with responses.RequestsMock():
scim_synchronizer.add_user_to_group(access.team, access.user)
assert len(responses.calls) == 0
@mock.patch.object(Logger, "info")
def test_utils_webhooks_add_user_to_group_success(mock_info):
"""The user passed to the function should get added."""
identity = factories.IdentityFactory()
access = factories.TeamAccessFactory(user=identity.user)
webhooks = factories.TeamWebhookFactory.create_batch(2, team=access.team)
with responses.RequestsMock() as rsps:
# Ensure successful response by scim provider using "responses":
rsps.add(
rsps.PATCH,
re.compile(r".*/Groups/.*"),
body="{}",
status=200,
content_type="application/json",
)
scim_synchronizer.add_user_to_group(access.team, access.user)
for i, webhook in enumerate(webhooks):
assert rsps.calls[i].request.url == webhook.url
# Check headers
headers = rsps.calls[i].request.headers
assert "Authorization" not in headers
assert headers["Content-Type"] == "application/json"
# Payload sent to scim provider
for call in rsps.calls:
payload = json.loads(call.request.body)
assert payload == {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "add",
"path": "members",
"value": [
{
"value": str(access.user.id),
"email": identity.email,
"type": "User",
}
],
}
],
}
# Logger
assert mock_info.call_count == 2
for i, webhook in enumerate(webhooks):
assert mock_info.call_args_list[i][0] == (
"%s synchronization succeeded with %s",
"add_user_to_group",
webhook.url,
)
# Status
for webhook in webhooks:
webhook.refresh_from_db()
assert webhook.status == "success"
@mock.patch.object(Logger, "info")
def test_utils_webhooks_remove_user_from_group_success(mock_info):
"""The user passed to the function should get removed."""
identity = factories.IdentityFactory()
access = factories.TeamAccessFactory(user=identity.user)
webhooks = factories.TeamWebhookFactory.create_batch(2, team=access.team)
with responses.RequestsMock() as rsps:
# Ensure successful response by scim provider using "responses":
rsps.add(
rsps.PATCH,
re.compile(r".*/Groups/.*"),
body="{}",
status=200,
content_type="application/json",
)
scim_synchronizer.remove_user_from_group(access.team, access.user)
for i, webhook in enumerate(webhooks):
assert rsps.calls[i].request.url == webhook.url
# Payload sent to scim provider
for call in rsps.calls:
payload = json.loads(call.request.body)
assert payload == {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "remove",
"path": "members",
"value": [
{
"value": str(access.user.id),
"email": identity.email,
"type": "User",
}
],
}
],
}
# Logger
assert mock_info.call_count == 2
for i, webhook in enumerate(webhooks):
assert mock_info.call_args_list[i][0] == (
"%s synchronization succeeded with %s",
"remove_user_from_group",
webhook.url,
)
# Status
for webhook in webhooks:
webhook.refresh_from_db()
assert webhook.status == "success"
@mock.patch.object(Logger, "error")
@mock.patch.object(Logger, "info")
def test_utils_webhooks_add_user_to_group_failure(mock_info, mock_error):
"""The logger should be called on webhook call failure."""
identity = factories.IdentityFactory()
access = factories.TeamAccessFactory(user=identity.user)
webhooks = factories.TeamWebhookFactory.create_batch(2, team=access.team)
with responses.RequestsMock() as rsps:
# Simulate webhook failure using "responses":
rsps.add(
rsps.PATCH,
re.compile(r".*/Groups/.*"),
body="{}",
status=random.choice([404, 301, 302]),
content_type="application/json",
)
scim_synchronizer.add_user_to_group(access.team, access.user)
for i, webhook in enumerate(webhooks):
assert rsps.calls[i].request.url == webhook.url
# Payload sent to scim provider
for call in rsps.calls:
payload = json.loads(call.request.body)
assert payload == {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "add",
"path": "members",
"value": [
{
"value": str(access.user.id),
"email": identity.email,
"type": "User",
}
],
}
],
}
# Logger
assert not mock_info.called
assert mock_error.call_count == 2
for i, webhook in enumerate(webhooks):
assert mock_error.call_args_list[i][0] == (
"%s synchronization failed with %s",
"add_user_to_group",
webhook.url,
)
# Status
for webhook in webhooks:
webhook.refresh_from_db()
assert webhook.status == "failure"
@mock.patch.object(Logger, "error")
@mock.patch.object(Logger, "info")
def test_utils_webhooks_add_user_to_group_retries(mock_info, mock_error):
"""webhooks synchronization supports retries."""
identity = factories.IdentityFactory()
access = factories.TeamAccessFactory(user=identity.user)
webhook = factories.TeamWebhookFactory(team=access.team)
url = re.compile(r".*/Groups/.*")
with responses.RequestsMock() as rsps:
# Make webhook fail 3 times before succeeding using "responses"
all_rsps = [
rsps.add(rsps.PATCH, url, status=500, content_type="application/json"),
rsps.add(rsps.PATCH, url, status=500, content_type="application/json"),
rsps.add(rsps.PATCH, url, status=500, content_type="application/json"),
rsps.add(rsps.PATCH, url, status=200, content_type="application/json"),
]
scim_synchronizer.add_user_to_group(access.team, access.user)
for i in range(4):
assert all_rsps[i].call_count == 1
assert rsps.calls[i].request.url == webhook.url
payload = json.loads(rsps.calls[i].request.body)
assert payload == {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "add",
"path": "members",
"value": [
{
"value": str(access.user.id),
"email": identity.email,
"type": "User",
}
],
}
],
}
# Logger
assert not mock_error.called
assert mock_info.call_count == 1
assert mock_info.call_args_list[0][0] == (
"%s synchronization succeeded with %s",
"add_user_to_group",
webhook.url,
)
# Status
webhook.refresh_from_db()
assert webhook.status == "success"
@mock.patch.object(Logger, "error")
@mock.patch.object(Logger, "info")
def test_utils_synchronize_course_runs_max_retries_exceeded(mock_info, mock_error):
"""Webhooks synchronization has exceeded max retries and should get logged."""
identity = factories.IdentityFactory()
access = factories.TeamAccessFactory(user=identity.user)
webhook = factories.TeamWebhookFactory(team=access.team)
with responses.RequestsMock() as rsps:
# Simulate webhook temporary failure using "responses":
rsp = rsps.add(
rsps.PATCH,
re.compile(r".*/Groups/.*"),
body="{}",
status=random.choice([500, 502]),
content_type="application/json",
)
scim_synchronizer.add_user_to_group(access.team, access.user)
assert rsp.call_count == 5
assert rsps.calls[0].request.url == webhook.url
payload = json.loads(rsps.calls[0].request.body)
assert payload == {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "add",
"path": "members",
"value": [
{
"value": str(access.user.id),
"email": identity.email,
"type": "User",
}
],
}
],
}
# Logger
assert not mock_info.called
assert mock_error.call_count == 1
assert mock_error.call_args_list[0][0] == (
"%s synchronization failed due to max retries exceeded with url %s",
"add_user_to_group",
webhook.url,
)
# Status
webhook.refresh_from_db()
assert webhook.status == "failure"
def test_utils_webhooks_add_user_to_group_authorization():
"""Secret token should be passed in authorization header when set."""
identity = factories.IdentityFactory()
access = factories.TeamAccessFactory(user=identity.user)
webhook = factories.TeamWebhookFactory(team=access.team, secret="123")
with responses.RequestsMock() as rsps:
# Ensure successful response by scim provider using "responses":
rsps.add(
rsps.PATCH,
re.compile(r".*/Groups/.*"),
body="{}",
status=200,
content_type="application/json",
)
scim_synchronizer.add_user_to_group(access.team, access.user)
assert rsps.calls[0].request.url == webhook.url
# Check headers
headers = rsps.calls[0].request.headers
assert headers["Authorization"] == "Bearer 123"
assert headers["Content-Type"] == "application/json"
# Status
webhook.refresh_from_db()
assert webhook.status == "success"
+25
View File
@@ -0,0 +1,25 @@
"""Utils for tests in the People core application"""
from rest_framework_simplejwt.tokens import AccessToken
class OIDCToken(AccessToken):
"""Set payload on token from user/contact/email"""
@classmethod
def for_user(cls, user):
"""Returns an authorization token for the given user for testing."""
identity = user.identities.filter(is_main=True).first()
token = cls()
token["first_name"] = (
user.profile_contact.short_name if user.profile_contact else "David"
)
token["last_name"] = (
" ".join(user.profile_contact.full_name.split()[1:])
if user.profile_contact
else "Bowman"
)
token["sub"] = str(identity.sub)
token["email"] = user.email
return token
+10
View File
@@ -0,0 +1,10 @@
"""Tokens for People's core app."""
from rest_framework_simplejwt.settings import api_settings
from rest_framework_simplejwt.tokens import Token
class BearerToken(Token):
"""Bearer token as emitted by Keycloak OIDC for example."""
token_type = "Bearer" # noqa: S105
lifetime = api_settings.ACCESS_TOKEN_LIFETIME
View File
-70
View File
@@ -1,70 +0,0 @@
"""A minimalist SCIM client to synchronize with remote service providers."""
import logging
import requests
from urllib3.util import Retry
logger = logging.getLogger(__name__)
adapter = requests.adapters.HTTPAdapter(
max_retries=Retry(
total=4,
backoff_factor=0.1,
status_forcelist=[500, 502],
allowed_methods=["PATCH"],
)
)
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)
class SCIMClient:
"""A minimalist SCIM client for our needs."""
def add_user_to_group(self, webhook, user):
"""Add a user to a group from its ID or email."""
payload = {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "add",
"path": "members",
"value": [
{"value": str(user.id), "email": user.email, "type": "User"}
],
}
],
}
return session.patch(
webhook.url,
json=payload,
headers=webhook.get_headers(),
verify=False,
timeout=3,
)
def remove_user_from_group(self, webhook, user):
"""Remove a user from a group by its ID or email."""
payload = {
"schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"],
"Operations": [
{
"op": "remove",
"path": "members",
"value": [
{"value": str(user.id), "email": user.email, "type": "User"}
],
}
],
}
return session.patch(
webhook.url,
json=payload,
headers=webhook.get_headers(),
verify=False,
timeout=3,
)
-75
View File
@@ -1,75 +0,0 @@
"""Fire webhooks with synchronous retries"""
import logging
import requests
from core.enums import WebhookStatusChoices
from .scim import SCIMClient
logger = logging.getLogger(__name__)
class WebhookSCIMClient:
"""Wraps the SCIM client to record call results on webhooks."""
def __getattr__(self, name):
"""Handle calls from webhooks to synchronize a team access with a distant application."""
def wrapper(team, user):
"""
Wrap SCIMClient calls to handle retries, error handling and storing result in the
calling Webhook instance.
"""
for webhook in team.webhooks.all():
if not webhook.url:
continue
client = SCIMClient()
status = WebhookStatusChoices.FAILURE
try:
response = getattr(client, name)(webhook, user)
except requests.exceptions.RetryError as exc:
logger.error(
"%s synchronization failed due to max retries exceeded with url %s",
name,
webhook.url,
exc_info=exc,
)
except requests.exceptions.RequestException as exc:
logger.error(
"%s synchronization failed with %s.",
name,
webhook.url,
exc_info=exc,
)
else:
extra = {
"response": response.content,
}
# pylint: disable=no-member
if response.status_code == requests.codes.ok:
logger.info(
"%s synchronization succeeded with %s",
name,
webhook.url,
extra=extra,
)
status = WebhookStatusChoices.SUCCESS
else:
logger.error(
"%s synchronization failed with %s",
name,
webhook.url,
extra=extra,
)
webhook._meta.model.objects.filter(id=webhook.id).update(status=status) # noqa
return wrapper
scim_synchronizer = WebhookSCIMClient()
-21
View File
@@ -1,21 +0,0 @@
"""Debug Urls to check the layout of emails"""
from django.urls import path
from .views import (
DebugViewHtml,
DebugViewTxt,
)
urlpatterns = [
path(
"__debug__/mail/invitation_html",
DebugViewHtml.as_view(),
name="debug.mail.invitation_html",
),
path(
"__debug__/mail/invitation_txt",
DebugViewTxt.as_view(),
name="debug.mail.invitation_txt",
),
]
-27
View File
@@ -1,27 +0,0 @@
"""Debug Views to check the layout of emails"""
from django.views.generic.base import TemplateView
class DebugBaseView(TemplateView):
"""Debug View to check the layout of emails"""
def get_context_data(self, **kwargs):
"""Generates sample datas to have a valid debug email"""
context = super().get_context_data(**kwargs)
context["title"] = "Development email preview"
return context
class DebugViewHtml(DebugBaseView):
"""Debug View for HTML Email Layout"""
template_name = "mail/html/invitation.html"
class DebugViewTxt(DebugBaseView):
"""Debug View for Text Email Layout"""
template_name = "mail/text/invitation.txt"
+1
View File
@@ -3,5 +3,6 @@
NB_OBJECTS = {
"users": 1000,
"teams": 100,
"max_identities_per_user": 3,
"max_users_per_team": 100,
}
@@ -11,14 +11,10 @@ from django import db
from django.conf import settings
from django.core.management.base import BaseCommand, CommandError
from faker import Faker
from core import models
from demo import defaults
fake = Faker()
logger = logging.getLogger("people.commands.demo.create_demo")
@@ -115,7 +111,7 @@ def create_demo(stdout):
for i in range(defaults.NB_OBJECTS["users"]):
queue.push(
models.User(
admin_email=f"user{i:d}@example.com",
email=f"user{i:d}@example.com",
password="!",
is_superuser=False,
is_active=True,
@@ -126,22 +122,18 @@ def create_demo(stdout):
queue.flush()
with Timeit(stdout, "Creating identities"):
users_values = list(models.User.objects.values("id", "admin_email"))
users_values = list(models.User.objects.values("id", "email"))
for user_dict in users_values:
for i in range(
random.choices(range(5), weights=[5, 50, 30, 10, 5], k=1)[0]
random.randint(0, defaults.NB_OBJECTS["max_identities_per_user"])
):
user_email = user_dict["admin_email"]
user_email = user_dict["email"]
queue.push(
models.Identity(
user_id=user_dict["id"],
sub=uuid4(),
email=f"identity{i:d}{user_email:s}",
is_main=(i == 0),
# Leave 3% of emails and names empty
email=f"identity{i:d}{user_email:s}"
if random.random() < 0.97
else None,
name=fake.name() if random.random() < 0.97 else None,
)
)
queue.flush()
@@ -2,7 +2,6 @@
Management command overriding the "createsuperuser" command to allow creating users
with their email and no username.
"""
from django.contrib.auth import get_user_model
from django.core.management.base import BaseCommand
@@ -17,7 +16,7 @@ class Command(BaseCommand):
def add_arguments(self, parser):
"""Define required arguments "email" and "password"."""
parser.add_argument(
"--admin_email",
"--email",
help=("Email for the user."),
)
parser.add_argument(
@@ -30,11 +29,11 @@ class Command(BaseCommand):
Given an email and a password, create a superuser or upgrade the existing
user to superuser status.
"""
email = options.get("admin_email")
email = options.get("email")
try:
user = User.objects.get(admin_email=email)
user = User.objects.get(email=email)
except User.DoesNotExist:
user = User(admin_email=email)
user = User(email=email)
message = "Superuser created successfully."
else:
if user.is_superuser and user.is_staff:
@@ -1,5 +1,4 @@
"""Test the `create_demo` management command"""
from unittest import mock
from django.core.management import call_command
@@ -31,14 +30,3 @@ def test_commands_create_demo():
assert models.Identity.objects.exists()
assert models.Team.objects.count() == 3
assert models.TeamAccess.objects.count() >= 3
def test_commands_createsuperuser():
"""
The createsuperuser management command should create a user
with superuser permissions.
"""
call_command("createsuperuser")
assert models.User.objects.count() == 1
+52 -199
View File
@@ -2,7 +2,7 @@ msgid ""
msgstr ""
"Project-Id-Version: lasuite-people\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2024-03-21 19:26+0000\n"
"POT-Creation-Date: 2024-01-03 23:15+0000\n"
"PO-Revision-Date: 2024-01-03 23:15\n"
"Last-Translator: \n"
"Language-Team: French\n"
@@ -17,322 +17,175 @@ msgstr ""
"X-Crowdin-File: backend.pot\n"
"X-Crowdin-File-ID: 2\n"
#: core/admin.py:70
msgid "Personal info"
msgstr ""
#: core/admin.py:72
msgid "Permissions"
msgstr ""
#: core/admin.py:84
msgid "Important dates"
msgstr ""
#: core/authentication.py:81
msgid "User info contained no recognizable user identification"
msgstr ""
#: core/authentication.py:114
msgid "Claims contained no recognizable user identification"
msgstr ""
#: core/models.py:38
#: core/models.py:30
msgid "Member"
msgstr ""
#: core/models.py:39
#: core/models.py:31
msgid "Administrator"
msgstr ""
#: core/models.py:40
#: core/models.py:32
msgid "Owner"
msgstr ""
#: core/models.py:52
#: core/models.py:44
msgid "id"
msgstr ""
#: core/models.py:53
#: core/models.py:45
msgid "primary key for the record as UUID"
msgstr ""
#: core/models.py:59
#: core/models.py:51
msgid "created at"
msgstr ""
#: core/models.py:60
#: core/models.py:52
msgid "date and time at which a record was created"
msgstr ""
#: core/models.py:65
#: core/models.py:57
msgid "updated at"
msgstr ""
#: core/models.py:66
#: core/models.py:58
msgid "date and time at which a record was last updated"
msgstr ""
#: core/models.py:97
#: core/models.py:89
msgid "full name"
msgstr ""
#: core/models.py:98
#: core/models.py:90
msgid "short name"
msgstr ""
#: core/models.py:103
#: core/models.py:95
msgid "contact information"
msgstr ""
#: core/models.py:104
#: core/models.py:96
msgid "A JSON object containing the contact information"
msgstr ""
#: core/models.py:118
#: core/models.py:110
msgid "contact"
msgstr ""
#: core/models.py:119
#: core/models.py:111
msgid "contacts"
msgstr ""
#: core/models.py:160 core/models.py:263 core/models.py:483
msgid "email address"
msgstr ""
#: core/models.py:172
#: core/models.py:173
msgid "language"
msgstr ""
#: core/models.py:173
#: core/models.py:174
msgid "The language in which the user wants to see the interface."
msgstr ""
#: core/models.py:179
#: core/models.py:180
msgid "The timezone in which the user wants to see times."
msgstr ""
#: core/models.py:182
#: core/models.py:183
msgid "device"
msgstr ""
#: core/models.py:184
#: core/models.py:185
msgid "Whether the user is a device or a real user."
msgstr ""
#: core/models.py:187
#: core/models.py:188
msgid "staff status"
msgstr ""
#: core/models.py:189
#: core/models.py:190
msgid "Whether the user can log into this admin site."
msgstr ""
#: core/models.py:192
#: core/models.py:193
msgid "active"
msgstr ""
#: core/models.py:195
msgid ""
"Whether this user should be treated as active. Unselect this instead of "
"deleting accounts."
msgstr ""
#: core/models.py:207
msgid "user"
#: core/models.py:196
msgid "Whether this user should be treated as active. Unselect this instead of deleting accounts."
msgstr ""
#: core/models.py:208
msgid "user"
msgstr ""
#: core/models.py:209
msgid "users"
msgstr ""
#: core/models.py:248
msgid ""
"Enter a valid sub. This value may contain only letters, numbers, and @/./+/-/"
"_ characters."
#: core/models.py:245
msgid "Enter a valid sub. This value may contain only letters, numbers, and @/./+/-/_ characters."
msgstr ""
#: core/models.py:255
#: core/models.py:252
msgid "sub"
msgstr ""
#: core/models.py:257
msgid ""
"Required. 255 characters or fewer. Letters, numbers, and @/./+/-/_ "
"characters only."
#: core/models.py:254
msgid "Required. 255 characters or fewer. Letters, numbers, and @/./+/-/_ characters only."
msgstr ""
#: core/models.py:264
msgid "name"
#: core/models.py:260
msgid "email address"
msgstr ""
#: core/models.py:266
#: core/models.py:262
msgid "main"
msgstr ""
#: core/models.py:268
#: core/models.py:264
msgid "Designates whether the email is the main one."
msgstr ""
#: core/models.py:274
#: core/models.py:270
msgid "identity"
msgstr ""
#: core/models.py:275
#: core/models.py:271
msgid "identities"
msgstr ""
#: core/models.py:282
#: core/models.py:278
msgid "This email address is already declared for this user."
msgstr ""
#: core/models.py:356
#: core/models.py:323
msgid "Team"
msgstr ""
#: core/models.py:357
#: core/models.py:324
msgid "Teams"
msgstr ""
#: core/models.py:417
#: core/models.py:375
msgid "Team/user relation"
msgstr ""
#: core/models.py:418
#: core/models.py:376
msgid "Team/user relations"
msgstr ""
#: core/models.py:423
#: core/models.py:381
msgid "This user is already in this team."
msgstr ""
#: core/models.py:500
msgid "Team invitation"
#: core/models.py:451
msgid "Token contained no recognizable user identification"
msgstr ""
#: core/models.py:501
msgid "Team invitations"
msgstr ""
#: core/models.py:526
msgid "This email is already associated to a registered user."
msgstr ""
#: core/models.py:568 core/models.py:573
msgid "Invitation to join Desk!"
msgstr ""
#: core/templates/mail/html/invitation.html:160
#: core/templates/mail/text/invitation.txt:3
msgid "La Suite Numérique"
msgstr ""
#: core/templates/mail/html/invitation.html:190
#: core/templates/mail/text/invitation.txt:5
msgid "Invitation to join a team"
msgstr ""
#: core/templates/mail/html/invitation.html:198
#: core/templates/mail/text/invitation.txt:8
msgid "Welcome to"
msgstr ""
#: core/templates/mail/html/invitation.html:216
#: core/templates/mail/text/invitation.txt:12
msgid "Logo"
msgstr ""
#: core/templates/mail/html/invitation.html:226
#: core/templates/mail/text/invitation.txt:14
msgid ""
"We are delighted to welcome you to our community on Equipes, your new "
"companion to simplify the management of your groups efficiently, "
"intuitively, and securely."
msgstr ""
#: core/templates/mail/html/invitation.html:231
#: core/templates/mail/text/invitation.txt:15
msgid ""
"Our application is designed to help you organize, collaborate, and manage "
"permissions."
msgstr ""
#: core/templates/mail/html/invitation.html:236
#: core/templates/mail/text/invitation.txt:16
msgid "With Equipes, you will be able to:"
msgstr ""
#: core/templates/mail/html/invitation.html:237
#: core/templates/mail/text/invitation.txt:17
msgid "Create customized groups according to your specific needs."
msgstr ""
#: core/templates/mail/html/invitation.html:238
#: core/templates/mail/text/invitation.txt:18
msgid "Invite members of your team or community in just a few clicks."
msgstr ""
#: core/templates/mail/html/invitation.html:239
#: core/templates/mail/text/invitation.txt:19
msgid ""
"Plan events, meetings, or activities effortlessly with our integrated "
"calendar."
msgstr ""
#: core/templates/mail/html/invitation.html:240
#: core/templates/mail/text/invitation.txt:20
msgid "Share documents, photos, and important information securely."
msgstr ""
#: core/templates/mail/html/invitation.html:241
#: core/templates/mail/text/invitation.txt:21
msgid ""
"Facilitate exchanges and communication with our messaging and group "
"discussion tools."
msgstr ""
#: core/templates/mail/html/invitation.html:252
#: core/templates/mail/text/invitation.txt:23
msgid "Visit Equipes"
msgstr ""
#: core/templates/mail/html/invitation.html:261
#: core/templates/mail/text/invitation.txt:25
msgid ""
"We are confident that Equipes will help you increase efficiency and "
"productivity while strengthening the bond among members."
msgstr ""
#: core/templates/mail/html/invitation.html:266
#: core/templates/mail/text/invitation.txt:26
msgid ""
"Feel free to explore all the features of the application and share your "
"feedback and suggestions with us. Your feedback is valuable to us and will "
"enable us to continually improve our service."
msgstr ""
#: core/templates/mail/html/invitation.html:271
#: core/templates/mail/text/invitation.txt:27
msgid ""
"Once again, welcome aboard! We are eager to accompany you on this group "
"management adventure."
msgstr ""
#: core/templates/mail/html/invitation.html:278
#: core/templates/mail/text/invitation.txt:29
msgid "Sincerely,"
msgstr ""
#: core/templates/mail/html/invitation.html:279
#: core/templates/mail/text/invitation.txt:31
msgid "The La Suite Numérique Team"
msgstr ""
#: people/settings.py:133
#: people/settings.py:132
msgid "English"
msgstr ""
#: people/settings.py:134
#: people/settings.py:133
msgid "French"
msgstr ""
-33
View File
@@ -1,33 +0,0 @@
"""Admin classes and registrations for People's mailbox manager app."""
from django.contrib import admin
from django.utils.translation import gettext_lazy as _
from mailbox_manager import models
@admin.register(models.MailDomain)
class MailDomainAdmin(admin.ModelAdmin):
"""Mail domain admin interface declaration."""
list_display = (
"name",
"created_at",
"updated_at",
)
search_fields = ("name",)
readonly_fields = ["created_at"]
@admin.register(models.MailDomainAccess)
class MailDomainAccessAdmin(admin.ModelAdmin):
"""Admin for mail domain accesses model."""
list_display = ("user", "domain")
@admin.register(models.Mailbox)
class MailboxAdmin(admin.ModelAdmin):
"""Admin for mailbox model."""
list_display = ("__str__", "domain")
@@ -1,12 +0,0 @@
"""Permission handlers for the People mailbox manager app."""
from core.api import permissions as core_permissions
class AccessPermission(core_permissions.IsAuthenticated):
"""Permission class for access objects."""
def has_object_permission(self, request, view, obj):
"""Check permission for a given object."""
abilities = obj.get_abilities(request.user)
return abilities.get(request.method.lower(), False)
@@ -1,42 +0,0 @@
"""Client serializers for People's mailbox manager app."""
from rest_framework import serializers
from mailbox_manager import models
class MailboxSerializer(serializers.ModelSerializer):
"""Serialize mailbox."""
class Meta:
model = models.Mailbox
fields = ["id", "local_part", "secondary_email"]
class MailDomainSerializer(serializers.ModelSerializer):
"""Serialize mail domain."""
class Meta:
model = models.MailDomain
lookup_field = "slug"
fields = [
"id",
"name",
"created_at",
"updated_at",
]
class MailDomainAccessSerializer(serializers.ModelSerializer):
"""Serialize mail domain accesses."""
class Meta:
model = models.MailDomainAccess
fields = [
"id",
"user",
"role",
"created_at",
"updated_at",
]
read_only_fields = ["id"]
-101
View File
@@ -1,101 +0,0 @@
"""API endpoints"""
from rest_framework import filters, mixins, viewsets
from rest_framework import permissions as drf_permissions
from core import models as core_models
from mailbox_manager import models
from mailbox_manager.api import permissions, serializers
# pylint: disable=too-many-ancestors
class MailDomainViewSet(
mixins.CreateModelMixin,
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.DestroyModelMixin,
viewsets.GenericViewSet,
):
"""
MailDomain viewset.
GET /api/<version>/mail-domains/
Return a list of mail domains user has access to.
GET /api/<version>/mail-domains/<domain-slug>/
Return details for a mail domain user has access to.
POST /api/<version>/mail-domains/ with expected data:
- name: str
Return newly created domain
DELETE /api/<version>/mail-domains/<domain-slug>/
Delete targeted team access
"""
permission_classes = [permissions.AccessPermission]
serializer_class = serializers.MailDomainSerializer
filter_backends = [filters.OrderingFilter]
ordering_fields = ["created_at", "name"]
ordering = ["-created_at"]
lookup_field = "slug"
queryset = models.MailDomain.objects.all()
def get_queryset(self):
return self.queryset.filter(accesses__user=self.request.user)
def perform_create(self, serializer):
"""Set the current user as owner of the newly created mail domain."""
domain = serializer.save()
models.MailDomainAccess.objects.create(
user=self.request.user,
domain=domain,
role=core_models.RoleChoices.OWNER,
)
# pylint: disable=too-many-ancestors
class MailDomainAccessViewSet(
mixins.ListModelMixin,
viewsets.GenericViewSet,
):
"""
MailDomainAccess viewset.
"""
permission_classes = [drf_permissions.IsAuthenticated]
serializer_class = serializers.MailDomainAccessSerializer
filter_backends = [filters.OrderingFilter]
ordering_fields = ["created_at", "user", "domain", "role"]
ordering = ["-created_at"]
queryset = models.MailDomainAccess.objects.all()
class MailBoxViewSet(
mixins.CreateModelMixin,
mixins.ListModelMixin,
viewsets.GenericViewSet,
):
"""MailBox ViewSet"""
permission_classes = [drf_permissions.IsAuthenticated]
serializer_class = serializers.MailboxSerializer
queryset = models.Mailbox.objects.all()
def get_queryset(self):
"""Custom queryset to get mailboxes related to a mail domain."""
domain_slug = self.kwargs.get("domain_slug", "")
if domain_slug:
return self.queryset.filter(domain__slug=domain_slug)
return self.queryset
def perform_create(self, serializer):
"""Create new mailbox."""
domain_slug = self.kwargs.get("domain_slug", "")
if domain_slug:
serializer.validated_data["domain"] = models.MailDomain.objects.get(
slug=domain_slug
)
super().perform_create(serializer)
-1
View File
@@ -1 +0,0 @@
"""People additionnal application, to manage email adresses."""
-67
View File
@@ -1,67 +0,0 @@
"""
Mailbox manager application factories
"""
from django.utils.text import slugify
import factory.fuzzy
from faker import Faker
from core import factories as core_factories
from core import models as core_models
from mailbox_manager import models
fake = Faker()
class MailDomainFactory(factory.django.DjangoModelFactory):
"""A factory to create mail domain."""
class Meta:
model = models.MailDomain
django_get_or_create = ("name",)
skip_postgeneration_save = True
name = factory.Faker("domain_name")
slug = factory.LazyAttribute(lambda o: slugify(o.name))
@factory.post_generation
def users(self, create, extracted, **kwargs):
"""Add users to domain from a given list of users with or without roles."""
if not create or not extracted:
return
for user_entry in extracted:
if isinstance(user_entry, core_models.User):
MailDomainAccessFactory(domain=self, user=user_entry)
else:
MailDomainAccessFactory(
domain=self, user=user_entry[0], role=user_entry[1]
)
class MailDomainAccessFactory(factory.django.DjangoModelFactory):
"""A factory to create mail domain accesses."""
class Meta:
model = models.MailDomainAccess
user = factory.SubFactory(core_factories.UserFactory)
domain = factory.SubFactory(MailDomainFactory)
role = factory.fuzzy.FuzzyChoice([r[0] for r in core_models.RoleChoices.choices])
class MailboxFactory(factory.django.DjangoModelFactory):
"""A factory to create mailboxes for mail domain members."""
class Meta:
model = models.Mailbox
class Params:
"""Parameters for fields."""
full_name = factory.Faker("name")
local_part = factory.LazyAttribute(lambda a: a.full_name.lower().replace(" ", "."))
domain = factory.SubFactory(MailDomainFactory)
secondary_email = factory.Faker("email")
@@ -1,67 +0,0 @@
# Generated by Django 5.0.3 on 2024-04-16 12:51
import django.core.validators
import django.db.models.deletion
import uuid
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='MailDomain',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('name', models.CharField(max_length=150, unique=True, verbose_name='name')),
],
options={
'verbose_name': 'Mail domain',
'verbose_name_plural': 'Mail domains',
'db_table': 'people_mail_domain',
},
),
migrations.CreateModel(
name='Mailbox',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('local_part', models.CharField(max_length=150, validators=[django.core.validators.RegexValidator(regex='^[a-zA-Z0-9_.+-]+$')], verbose_name='local_part')),
('secondary_email', models.EmailField(max_length=254, verbose_name='secondary email address')),
('domain', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='mail_domain', to='mailbox_manager.maildomain')),
],
options={
'verbose_name': 'Mailbox',
'verbose_name_plural': 'Mailboxes',
'db_table': 'people_mail_box',
'unique_together': {('local_part', 'domain')},
},
),
migrations.CreateModel(
name='MailDomainAccess',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created at')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated at')),
('role', models.CharField(choices=[('member', 'Member'), ('administrator', 'Administrator'), ('owner', 'Owner')], default='member', max_length=20)),
('domain', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='mail_domain_accesses', to='mailbox_manager.maildomain')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='mail_domain_accesses', to=settings.AUTH_USER_MODEL)),
],
options={
'verbose_name': 'User/mail domain relation',
'verbose_name_plural': 'User/mail domain relations',
'db_table': 'people_mail_domain_accesses',
'unique_together': {('user', 'domain')},
},
),
]
@@ -1,19 +0,0 @@
# Generated by Django 5.0.3 on 2024-04-17 11:58
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('mailbox_manager', '0001_initial'),
]
operations = [
migrations.AlterField(
model_name='maildomainaccess',
name='domain',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='accesses', to='mailbox_manager.maildomain'),
),
]
@@ -1,18 +0,0 @@
# Generated by Django 5.0.6 on 2024-06-03 14:14
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('mailbox_manager', '0002_alter_maildomainaccess_domain'),
]
operations = [
migrations.AddField(
model_name='maildomain',
name='slug',
field=models.SlugField(blank=True, max_length=80),
),
]
@@ -1,23 +0,0 @@
# Generated by Django 5.0.6 on 2024-06-03 14:14
from django.db import migrations, models
from django.utils.text import slugify
def populate_slug(apps, schema_editor):
MailDomain = apps.get_model('mailbox_manager', 'MailDomain')
for mail_domain in MailDomain.objects.filter(slug=''):
mail_domain.slug = slugify(mail_domain.name)
mail_domain.save()
class Migration(migrations.Migration):
dependencies = [
('mailbox_manager', '0003_maildomain_slug'),
]
operations = [
migrations.RunPython(populate_slug, reverse_code=migrations.RunPython.noop),
]
@@ -1,18 +0,0 @@
# Generated by Django 5.0.6 on 2024-06-03 14:20
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('mailbox_manager', '0004_maildomain_fill_slug'),
]
operations = [
migrations.AlterField(
model_name='maildomain',
name='slug',
field=models.SlugField(max_length=80, unique=True),
),
]
-118
View File
@@ -1,118 +0,0 @@
"""
Declare and configure the models for the People additional application : mailbox_manager
"""
from django.conf import settings
from django.core import validators
from django.db import models
from django.utils.text import slugify
from django.utils.translation import gettext_lazy as _
from core.models import BaseModel, RoleChoices
class MailDomain(BaseModel):
"""Domain names from which we will create email addresses (mailboxes)."""
name = models.CharField(
_("name"), max_length=150, null=False, blank=False, unique=True
)
slug = models.SlugField(null=False, blank=False, unique=True, max_length=80)
class Meta:
db_table = "people_mail_domain"
verbose_name = _("Mail domain")
verbose_name_plural = _("Mail domains")
def __str__(self):
return self.name
def save(self, *args, **kwargs):
if not self.slug:
self.slug = slugify(self.name)
return super().save(*args, **kwargs)
def get_abilities(self, user):
"""
Compute and return abilities for a given user on the domain.
"""
is_owner_or_admin = False
role = None
if user.is_authenticated:
try:
role = self.accesses.filter(user=user).values("role")[0]["role"]
except (MailDomainAccess.DoesNotExist, IndexError):
role = None
is_owner_or_admin = role in [RoleChoices.OWNER, RoleChoices.ADMIN]
return {
"get": bool(role),
"patch": is_owner_or_admin,
"put": is_owner_or_admin,
"delete": role == RoleChoices.OWNER,
"manage_accesses": is_owner_or_admin,
}
class MailDomainAccess(BaseModel):
"""Allow to manage users' accesses to mail domains."""
user = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name="mail_domain_accesses",
null=False,
blank=False,
)
domain = models.ForeignKey(
MailDomain,
on_delete=models.CASCADE,
related_name="accesses",
null=False,
blank=False,
)
role = models.CharField(
max_length=20, choices=RoleChoices.choices, default=RoleChoices.MEMBER
)
class Meta:
db_table = "people_mail_domain_accesses"
verbose_name = _("User/mail domain relation")
verbose_name_plural = _("User/mail domain relations")
unique_together = ("user", "domain")
def __str__(self):
return f"Access of user {self.user} on domain {self.domain}."
class Mailbox(BaseModel):
"""Mailboxes for users from mail domain."""
local_part = models.CharField(
_("local_part"),
max_length=150,
null=False,
blank=False,
validators=[validators.RegexValidator(regex="^[a-zA-Z0-9_.+-]+$")],
)
domain = models.ForeignKey(
MailDomain,
on_delete=models.CASCADE,
related_name="mail_domain",
null=False,
blank=False,
)
secondary_email = models.EmailField(
_("secondary email address"), null=False, blank=False
)
class Meta:
db_table = "people_mail_box"
verbose_name = _("Mailbox")
verbose_name_plural = _("Mailboxes")
unique_together = ("local_part", "domain")
def __str__(self):
return f"{self.local_part!s}@{self.domain.name:s}"
@@ -1,73 +0,0 @@
"""
Tests for MailDomains API endpoint in People's app mailbox_manager. Focus on "create" action.
"""
import pytest
from rest_framework import status
from rest_framework.test import APIClient
from core import factories as core_factories
from mailbox_manager import factories, models
pytestmark = pytest.mark.django_db
def test_api_mail_domains__create_anonymous():
"""Anonymous users should not be allowed to create mail domains."""
response = APIClient().post(
"/api/v1.0/mail-domains/",
{
"name": "mydomain.com",
},
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert not models.MailDomain.objects.exists()
def test_api_mail_domains__create_name_unique():
"""
Creating domain should raise an error if already existing name.
"""
factories.MailDomainFactory(name="existing_domain.com")
identity = core_factories.IdentityFactory()
client = APIClient()
client.force_login(identity.user)
response = client.post(
"/api/v1.0/mail-domains/",
{
"name": "existing_domain.com",
},
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["name"] == ["Mail domain with this name already exists."]
def test_api_mail_domains__create_authenticated():
"""
Authenticated users should be able to create mail domains
and should automatically be added as owner of the newly created domain.
"""
identity = core_factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(identity.user)
response = client.post(
"/api/v1.0/mail-domains/",
{
"name": "mydomain.com",
},
format="json",
)
assert response.status_code == status.HTTP_201_CREATED
domain = models.MailDomain.objects.get()
assert domain.name == "mydomain.com"
assert domain.accesses.filter(role="owner", user=user).exists()
@@ -1,107 +0,0 @@
"""
Tests for MailDomains API endpoint, in People's mailbox manager app. Focus on "delete" action.
"""
import pytest
from rest_framework import status
from rest_framework.test import APIClient
from core import factories as core_factories
from mailbox_manager import factories, models
pytestmark = pytest.mark.django_db
def test_api_mail_domains__delete_anonymous():
"""Anonymous users should not be allowed to destroy a team."""
domain = factories.MailDomainFactory()
response = APIClient().delete(
f"/api/v1.0/mail-domains/{domain.slug}/",
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert models.MailDomain.objects.count() == 1
def test_api_mail_domains__delete_authenticated_unrelated():
"""
Authenticated users should not be allowed to delete a domain to which they are not
related.
"""
identity = core_factories.IdentityFactory()
domain = factories.MailDomainFactory()
client = APIClient()
client.force_login(identity.user)
response = client.delete(
f"/api/v1.0/mail-domains/{domain.slug}/",
)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json() == {"detail": "No MailDomain matches the given query."}
assert models.MailDomain.objects.count() == 1
def test_api_mail_domains__delete_authenticated_member():
"""
Authenticated users should not be allowed to delete a domain
to which they are only a member.
"""
identity = core_factories.IdentityFactory()
user = identity.user
domain = factories.MailDomainFactory(users=[(user, "member")])
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/mail-domains/{domain.slug}/",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert response.json() == {
"detail": "You do not have permission to perform this action."
}
assert models.MailDomain.objects.count() == 1
def test_api_mail_domains__delete_authenticated_administrator():
"""
Authenticated users should not be allowed to delete a domain
for which they are administrator.
"""
identity = core_factories.IdentityFactory()
user = identity.user
domain = factories.MailDomainFactory(users=[(user, "administrator")])
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/mail-domains/{domain.slug}/",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert response.json() == {
"detail": "You do not have permission to perform this action."
}
assert models.MailDomain.objects.count() == 1
def test_api_mail_domains__delete_authenticated_owner():
"""
Authenticated users should be able to delete a domain
for which they are directly owner.
"""
identity = core_factories.IdentityFactory()
user = identity.user
domain = factories.MailDomainFactory(users=[(user, "owner")])
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/mail-domains/{domain.slug}/",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert models.MailDomain.objects.exists() is False
@@ -1,56 +0,0 @@
"""
Tests for MailDomains API endpoint in People's mailbox manager app. Focus on "list" action.
"""
import pytest
from rest_framework import status
from rest_framework.test import APIClient
from core import factories as core_factories
from mailbox_manager import factories
pytestmark = pytest.mark.django_db
def test_api_mail_domains__list_anonymous():
"""Anonymous users should not be allowed to list mail domains."""
factories.MailDomainFactory.create_batch(3)
response = APIClient().get("/api/v1.0/mail-domains/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_mail_domains__list_authenticated():
"""
Authenticated users should be able to list domains
to which they have access.
"""
identity = core_factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
expected_ids = {
str(access.domain.id)
for access in factories.MailDomainAccessFactory.create_batch(5, user=user)
}
factories.MailDomainFactory.create_batch(2) # Other teams
factories.MailDomainAccessFactory.create_batch(2) # Other teams and accesses
response = client.get(
"/api/v1.0/mail-domains/",
)
assert response.status_code == status.HTTP_200_OK
results = response.json()["results"]
assert len(results) == 5
results_id = {result["id"] for result in results}
assert expected_ids == results_id
@@ -1,71 +0,0 @@
"""
Tests for MailDomains API endpoint in People's mailbox manager app. Focus on "retrieve" action.
"""
import pytest
from rest_framework import status
from rest_framework.test import APIClient
from core import factories as core_factories
from mailbox_manager import factories
pytestmark = pytest.mark.django_db
def test_api_mail_domains__retrieve_anonymous():
"""Anonymous users should not be allowed to retrieve a domain."""
domain = factories.MailDomainFactory()
response = APIClient().get(f"/api/v1.0/mail-domains/{domain.slug}/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert response.json() == {
"detail": "Authentication credentials were not provided."
}
def test_api_mail_domains__retrieve_authenticated_unrelated():
"""
Authenticated users should not be allowed to retrieve a domain
to which they have access.
"""
identity = core_factories.IdentityFactory()
client = APIClient()
client.force_login(identity.user)
domain = factories.MailDomainFactory()
response = client.get(
f"/api/v1.0/mail-domains/{domain.slug}/",
)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json() == {"detail": "No MailDomain matches the given query."}
def test_api_mail_domains__retrieve_authenticated_related():
"""
Authenticated users should be allowed to retrieve a domain
to which they have access.
"""
identity = core_factories.IdentityFactory()
user = identity.user
client = APIClient()
client.force_login(user)
domain = factories.MailDomainFactory()
factories.MailDomainAccessFactory(domain=domain, user=user)
response = client.get(
f"/api/v1.0/mail-domains/{domain.slug}/",
)
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"id": str(domain.id),
"name": domain.name,
"created_at": domain.created_at.isoformat().replace("+00:00", "Z"),
"updated_at": domain.updated_at.isoformat().replace("+00:00", "Z"),
}

Some files were not shown because too many files have changed in this diff Show More