fix import
parent
8637766c76
commit
f53cc22cbc
|
@ -5,6 +5,8 @@ from selfprivacy_api.graphql.mutations.mutation_interface import (
|
||||||
)
|
)
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
|
|
||||||
|
from selfprivacy_api.utils import ReadUserData
|
||||||
|
|
||||||
|
|
||||||
@strawberry.enum
|
@strawberry.enum
|
||||||
class UserType(Enum):
|
class UserType(Enum):
|
||||||
|
@ -28,3 +30,36 @@ class UserMutationReturn(MutationReturnInterface):
|
||||||
"""Return type for user mutation"""
|
"""Return type for user mutation"""
|
||||||
|
|
||||||
user: typing.Optional[User]
|
user: typing.Optional[User]
|
||||||
|
|
||||||
|
def get_user_by_username(username):
|
||||||
|
with ReadUserData() as data:
|
||||||
|
|
||||||
|
if username == "root":
|
||||||
|
if data["ssh"]["rootKeys"] not in data:
|
||||||
|
data["ssh"]["rootKeys"] = []
|
||||||
|
|
||||||
|
return User(
|
||||||
|
user_type=UserType.ROOT,
|
||||||
|
username="root",
|
||||||
|
ssh_keys=data["ssh"]["rootKeys"],
|
||||||
|
)
|
||||||
|
elif username == data["username"]:
|
||||||
|
if "sshKeys" not in data:
|
||||||
|
data["sshKeys"] = []
|
||||||
|
|
||||||
|
return User(
|
||||||
|
user_type=UserType.PRIMARY,
|
||||||
|
username=username,
|
||||||
|
ssh_keys=data["sshKeys"],
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
for user in data["users"]:
|
||||||
|
if user["username"] == username:
|
||||||
|
if "sshKeys" not in user:
|
||||||
|
user["sshKeys"] = []
|
||||||
|
|
||||||
|
return User(
|
||||||
|
user_type=UserType.NORMAL,
|
||||||
|
username=username,
|
||||||
|
ssh_keys=user["sshKeys"],
|
||||||
|
)
|
||||||
|
|
|
@ -7,8 +7,6 @@ import os
|
||||||
import subprocess
|
import subprocess
|
||||||
import portalocker
|
import portalocker
|
||||||
|
|
||||||
from selfprivacy_api.graphql.common_types.user import User, UserType
|
|
||||||
|
|
||||||
|
|
||||||
USERDATA_FILE = "/etc/nixos/userdata/userdata.json"
|
USERDATA_FILE = "/etc/nixos/userdata/userdata.json"
|
||||||
TOKENS_FILE = "/etc/nixos/userdata/tokens.json"
|
TOKENS_FILE = "/etc/nixos/userdata/tokens.json"
|
||||||
|
@ -177,35 +175,3 @@ def hash_password(password):
|
||||||
return hashed_password
|
return hashed_password
|
||||||
|
|
||||||
|
|
||||||
def get_user_by_username(username):
|
|
||||||
with ReadUserData() as data:
|
|
||||||
|
|
||||||
if username == "root":
|
|
||||||
if data["ssh"]["rootKeys"] not in data:
|
|
||||||
data["ssh"]["rootKeys"] = []
|
|
||||||
|
|
||||||
return User(
|
|
||||||
user_type=UserType.ROOT,
|
|
||||||
username="root",
|
|
||||||
ssh_keys=data["ssh"]["rootKeys"],
|
|
||||||
)
|
|
||||||
elif username == data["username"]:
|
|
||||||
if "sshKeys" not in data:
|
|
||||||
data["sshKeys"] = []
|
|
||||||
|
|
||||||
return User(
|
|
||||||
user_type=UserType.PRIMARY,
|
|
||||||
username=username,
|
|
||||||
ssh_keys=data["sshKeys"],
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
for user in data["users"]:
|
|
||||||
if user["username"] == username:
|
|
||||||
if "sshKeys" not in user:
|
|
||||||
user["sshKeys"] = []
|
|
||||||
|
|
||||||
return User(
|
|
||||||
user_type=UserType.NORMAL,
|
|
||||||
username=username,
|
|
||||||
ssh_keys=user["sshKeys"],
|
|
||||||
)
|
|
||||||
|
|
|
@ -20,5 +20,9 @@ def generate_system_query(query_array):
|
||||||
return "query TestSystem {\n system {" + "\n".join(query_array) + "}\n}"
|
return "query TestSystem {\n system {" + "\n".join(query_array) + "}\n}"
|
||||||
|
|
||||||
|
|
||||||
|
def generate_users_query(query_array):
|
||||||
|
return "query TestUsers {\n users {" + "\n".join(query_array) + "}\n}"
|
||||||
|
|
||||||
|
|
||||||
def mnemonic_to_hex(mnemonic):
|
def mnemonic_to_hex(mnemonic):
|
||||||
return Mnemonic(language="english").to_entropy(mnemonic).hex()
|
return Mnemonic(language="english").to_entropy(mnemonic).hex()
|
||||||
|
|
|
@ -3,7 +3,7 @@
|
||||||
import json
|
import json
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from tests.common import generate_system_query, read_json, write_json
|
from tests.common import generate_system_query, generate_users_query, read_json, write_json
|
||||||
|
|
||||||
|
|
||||||
def read_json(file_path):
|
def read_json(file_path):
|
||||||
|
@ -129,7 +129,7 @@ def test_graphql_get_users_unauthorized(client, some_users, mock_subprocess_pope
|
||||||
response = client.get(
|
response = client.get(
|
||||||
"/graphql",
|
"/graphql",
|
||||||
json={
|
json={
|
||||||
"query": generate_system_query([API_USERS_INFO]),
|
"query": generate_users_query([API_USERS_INFO]),
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
|
@ -140,7 +140,7 @@ def test_graphql_get_some_users(authorized_client, some_users, mock_subprocess_p
|
||||||
response = authorized_client.get(
|
response = authorized_client.get(
|
||||||
"/graphql",
|
"/graphql",
|
||||||
json={
|
json={
|
||||||
"query": generate_system_query([API_USERS_INFO]),
|
"query": generate_users_query([API_USERS_INFO]),
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
|
|
Loading…
Reference in New Issue