charles/peernet
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add: demo server and client

Browse Source
This commit is contained in:
Charles
2024-04-06 20:45:15 -07:00
parent 5a149bc922
commit 72effb5b2a
100 changed files with 11603 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
demo_server/openapi_server/__init__.py Normal file
View File

19
demo_server/openapi_server/__main__.py Normal file
View File

@@ -0,0 +1,19 @@
#!/usr/bin/env python3
import connexion
from openapi_server import encoder
def main():
app = connexion.App(__name__, specification_dir='./openapi/')
app.app.json_encoder = encoder.JSONEncoder
app.add_api('openapi.yaml',
arguments={'title': 'PeerNetService API'},
pythonic_params=True)
app.run(port=8080)
if __name__ == '__main__':
main()

0
demo_server/openapi_server/controllers/__init__.py Normal file
View File

227
demo_server/openapi_server/controllers/peer_net_service_controller.py Normal file
View File

@@ -0,0 +1,227 @@
import connexion
from typing import Dict
from typing import Tuple
from typing import Union
from openapi_server.models.claim_ice_candidates_response import ClaimIceCandidatesResponse # noqa: E501
from openapi_server.models.ice_candidate import IceCandidate # noqa: E501
from openapi_server.models.knock import Knock # noqa: E501
from openapi_server.models.list_knocks_response import ListKnocksResponse # noqa: E501
from openapi_server.models.room import Room # noqa: E501
from openapi_server.models.server import Server # noqa: E501
from openapi_server.models.service import Service # noqa: E501
from openapi_server.models.status import Status # noqa: E501
from openapi_server import util
servers = {}
rooms = {}
knocks = {}
sessions = {}
def peer_net_service_claim_ice_candidates(session): # noqa: E501
"""peer_net_service_claim_ice_candidates
Acts as both List and Delete atomically. # noqa: E501
:param session: The session id.
:type session: str
:param name:
:type name: str
:rtype: Union[ClaimIceCandidatesResponse, Tuple[ClaimIceCandidatesResponse, int], Tuple[ClaimIceCandidatesResponse, int, Dict[str, str]]
"""
if session not in sessions:
return 'not found', 404
candidates = [sessions[session][k] for k in sessions[session]]
sessions[session] = {}
return ClaimIceCandidatesResponse(ice_candidates=candidates)
def peer_net_service_create_ice_candidate(session, ice_candidate=None, target_session=None): # noqa: E501
"""peer_net_service_create_ice_candidate
# noqa: E501
:param session: The session id.
:type session: str
:param ice_candidate:
:type ice_candidate: dict | bytes
:param target_session:
:type target_session: str
:rtype: Union[IceCandidate, Tuple[IceCandidate, int], Tuple[IceCandidate, int, Dict[str, str]]
"""
if connexion.request.is_json:
ice_candidate = IceCandidate.from_dict(connexion.request.get_json()) # noqa: E501
sessions[session][ice_candidate.name] = ice_candidate
return ice_candidate
def peer_net_service_create_knock(server, service, knock=None): # noqa: E501
"""peer_net_service_create_knock
Creates a knock that will be answered by the peer; the signaler may delete the knock, regardless of the state, when it ages. # noqa: E501
:param server: The server id.
:type server: str
:param service: The service id.
:type service: str
:param knock:
:type knock: dict | bytes
:rtype: Union[Knock, Tuple[Knock, int], Tuple[Knock, int, Dict[str, str]]
"""
if connexion.request.is_json:
knock = Knock.from_dict(connexion.request.get_json()) # noqa: E501
print(knock)
if server not in servers or service not in [s.name for s in servers[server].services]:
return 'not found', 404
knocks[server][service][knock.name] = knock
sessions[knock.offer.name] = {}
return knock
def peer_net_service_create_server(server=None): # noqa: E501
"""peer_net_service_create_server
# noqa: E501
:param server:
:type server: dict | bytes
:rtype: Union[Server, Tuple[Server, int], Tuple[Server, int, Dict[str, str]]
"""
if connexion.request.is_json:
server = Server.from_dict(connexion.request.get_json()) # noqa: E501
servers[server.name] = server
for room in server.rooms:
if room not in rooms:
rooms[room] = {}
rooms[room][server.name] = server
knocks[server.name] = {}
for service in server.services:
knocks[server.name][service.name] = {}
return server
def peer_net_service_create_service(server, service): # noqa: E501
"""peer_net_service_create_service
# noqa: E501
:param server: The server id.
:type server: str
:param service:
:type service: dict | bytes
:rtype: Union[Service, Tuple[Service, int], Tuple[Service, int, Dict[str, str]]
"""
if connexion.request.is_json:
service = Service.from_dict(connexion.request.get_json()) # noqa: E501
return 'do some magic!'
def peer_net_service_delete_server(name=None): # noqa: E501
"""peer_net_service_delete_server
# noqa: E501
:param name:
:type name: str
:rtype: Union[None, Tuple[None, int], Tuple[None, int, Dict[str, str]]
"""
return 'do some magic!'
def peer_net_service_delete_service(server, service): # noqa: E501
"""peer_net_service_delete_service
# noqa: E501
:param server: The server id.
:type server: str
:param service: The service id.
:type service: str
:rtype: Union[None, Tuple[None, int], Tuple[None, int, Dict[str, str]]
"""
return 'do some magic!'
def peer_net_service_get_knock(server, service, knock): # noqa: E501
"""peer_net_service_get_knock
# noqa: E501
:param server: The server id.
:type server: str
:param service: The service id.
:type service: str
:param knock: The knock id.
:type knock: str
:rtype: Union[Knock, Tuple[Knock, int], Tuple[Knock, int, Dict[str, str]]
"""
if server not in knocks or service not in knocks[server] or knock not in knocks[server][service]:
return 'not found', 404
return knocks[server][service][knock]
def peer_net_service_get_room(room): # noqa: E501
"""peer_net_service_get_room
# noqa: E501
:param room: The room id.
:type room: str
:rtype: Union[Room, Tuple[Room, int], Tuple[Room, int, Dict[str, str]]
"""
if room not in rooms:
return 'not found', 404
s = []
for server in rooms[room]:
s.append(servers[server])
return Room(name=room, servers=s)
def peer_net_service_list_knocks(server, service): # noqa: E501
"""peer_net_service_list_knocks
# noqa: E501
:param server: The server id.
:type server: str
:param service: The service id.
:type service: str
:rtype: Union[ListKnocksResponse, Tuple[ListKnocksResponse, int], Tuple[ListKnocksResponse, int, Dict[str, str]]
"""
if server not in knocks or service not in knocks[server]:
return 'not found', 404
return ListKnocksResponse(knocks=[knocks[server][service][k] for k in knocks[server][service]])
def peer_net_service_update_knock(server, service, knock, knock2=None): # noqa: E501
"""peer_net_service_update_knock
# noqa: E501
:param server: The server id.
:type server: str
:param service: The service id.
:type service: str
:param knock: The knock id.
:type knock: str
:param knock2:
:type knock2: dict | bytes
:rtype: Union[Knock, Tuple[Knock, int], Tuple[Knock, int, Dict[str, str]]
"""
if connexion.request.is_json:
knock2 = Knock.from_dict(connexion.request.get_json()) # noqa: E501
if server not in servers or service not in [s.name for s in servers[server].services]:
return 'not found', 404
knocks[server][service][knock] = knock2
sessions[knock2.answer.name] = {}
return knock2

2
demo_server/openapi_server/controllers/security_controller.py Normal file
View File

@@ -0,0 +1,2 @@
from typing import List

19
demo_server/openapi_server/encoder.py Normal file
View File

@@ -0,0 +1,19 @@
from connexion.apps.flask_app import FlaskJSONEncoder
from openapi_server.models.base_model import Model
class JSONEncoder(FlaskJSONEncoder):
include_nulls = False
def default(self, o):
if isinstance(o, Model):
dikt = {}
for attr in o.openapi_types:
value = getattr(o, attr)
if value is None and not self.include_nulls:
continue
attr = o.attribute_map[attr]
dikt[attr] = value
return dikt
return FlaskJSONEncoder.default(self, o)

12
demo_server/openapi_server/models/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
# flake8: noqa
# import models into model package
from openapi_server.models.claim_ice_candidates_response import ClaimIceCandidatesResponse
from openapi_server.models.google_protobuf_any import GoogleProtobufAny
from openapi_server.models.ice_candidate import IceCandidate
from openapi_server.models.ice_session_description import IceSessionDescription
from openapi_server.models.knock import Knock
from openapi_server.models.list_knocks_response import ListKnocksResponse
from openapi_server.models.room import Room
from openapi_server.models.server import Server
from openapi_server.models.service import Service
from openapi_server.models.status import Status

68
demo_server/openapi_server/models/base_model.py Normal file
View File

@@ -0,0 +1,68 @@
import pprint
import typing
from openapi_server import util
T = typing.TypeVar('T')
class Model:
# openapiTypes: The key is attribute name and the
# value is attribute type.
openapi_types: typing.Dict[str, type] = {}
# attributeMap: The key is attribute name and the
# value is json key in definition.
attribute_map: typing.Dict[str, str] = {}
@classmethod
def from_dict(cls: typing.Type[T], dikt) -> T:
"""Returns the dict as a model"""
return util.deserialize_model(dikt, cls)
def to_dict(self):
"""Returns the model properties as a dict
:rtype: dict
"""
result = {}
for attr in self.openapi_types:
value = getattr(self, attr)
if isinstance(value, list):
result[attr] = list(map(
lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
value
))
elif hasattr(value, "to_dict"):
result[attr] = value.to_dict()
elif isinstance(value, dict):
result[attr] = dict(map(
lambda item: (item[0], item[1].to_dict())
if hasattr(item[1], "to_dict") else item,
value.items()
))
else:
result[attr] = value
return result
def to_str(self):
"""Returns the string representation of the model
:rtype: str
"""
return pprint.pformat(self.to_dict())
def __repr__(self):
"""For `print` and `pprint`"""
return self.to_str()
def __eq__(self, other):
"""Returns true if both objects are equal"""
return self.__dict__ == other.__dict__
def __ne__(self, other):
"""Returns true if both objects are not equal"""
return not self == other

63
demo_server/openapi_server/models/claim_ice_candidates_response.py Normal file
View File

@@ -0,0 +1,63 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server.models.ice_candidate import IceCandidate
from openapi_server import util
from openapi_server.models.ice_candidate import IceCandidate # noqa: E501
class ClaimIceCandidatesResponse(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, ice_candidates=None): # noqa: E501
"""ClaimIceCandidatesResponse - a model defined in OpenAPI
:param ice_candidates: The ice_candidates of this ClaimIceCandidatesResponse. # noqa: E501
:type ice_candidates: List[IceCandidate]
"""
self.openapi_types = {
'ice_candidates': List[IceCandidate]
}
self.attribute_map = {
'ice_candidates': 'iceCandidates'
}
self._ice_candidates = ice_candidates
@classmethod
def from_dict(cls, dikt) -> 'ClaimIceCandidatesResponse':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The ClaimIceCandidatesResponse of this ClaimIceCandidatesResponse. # noqa: E501
:rtype: ClaimIceCandidatesResponse
"""
return util.deserialize_model(dikt, cls)
@property
def ice_candidates(self) -> List[IceCandidate]:
"""Gets the ice_candidates of this ClaimIceCandidatesResponse.
:return: The ice_candidates of this ClaimIceCandidatesResponse.
:rtype: List[IceCandidate]
"""
return self._ice_candidates
@ice_candidates.setter
def ice_candidates(self, ice_candidates: List[IceCandidate]):
"""Sets the ice_candidates of this ClaimIceCandidatesResponse.
:param ice_candidates: The ice_candidates of this ClaimIceCandidatesResponse.
:type ice_candidates: List[IceCandidate]
"""
self._ice_candidates = ice_candidates

63
demo_server/openapi_server/models/google_protobuf_any.py Normal file
View File

@@ -0,0 +1,63 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server import util
class GoogleProtobufAny(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, type=None): # noqa: E501
"""GoogleProtobufAny - a model defined in OpenAPI
:param type: The type of this GoogleProtobufAny. # noqa: E501
:type type: str
"""
self.openapi_types = {
'type': str
}
self.attribute_map = {
'type': '@type'
}
self._type = type
@classmethod
def from_dict(cls, dikt) -> 'GoogleProtobufAny':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The GoogleProtobufAny of this GoogleProtobufAny. # noqa: E501
:rtype: GoogleProtobufAny
"""
return util.deserialize_model(dikt, cls)
@property
def type(self) -> str:
"""Gets the type of this GoogleProtobufAny.
The type of the serialized message. # noqa: E501
:return: The type of this GoogleProtobufAny.
:rtype: str
"""
return self._type
@type.setter
def type(self, type: str):
"""Sets the type of this GoogleProtobufAny.
The type of the serialized message. # noqa: E501
:param type: The type of this GoogleProtobufAny.
:type type: str
"""
self._type = type

165
demo_server/openapi_server/models/ice_candidate.py Normal file
View File

@@ -0,0 +1,165 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server import util
class IceCandidate(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, name=None, candidate=None, sdp_mid=None, sdp_line_index=None, username_fragment=None): # noqa: E501
"""IceCandidate - a model defined in OpenAPI
:param name: The name of this IceCandidate. # noqa: E501
:type name: str
:param candidate: The candidate of this IceCandidate. # noqa: E501
:type candidate: str
:param sdp_mid: The sdp_mid of this IceCandidate. # noqa: E501
:type sdp_mid: str
:param sdp_line_index: The sdp_line_index of this IceCandidate. # noqa: E501
:type sdp_line_index: int
:param username_fragment: The username_fragment of this IceCandidate. # noqa: E501
:type username_fragment: str
"""
self.openapi_types = {
'name': str,
'candidate': str,
'sdp_mid': str,
'sdp_line_index': int,
'username_fragment': str
}
self.attribute_map = {
'name': 'name',
'candidate': 'candidate',
'sdp_mid': 'sdpMid',
'sdp_line_index': 'sdpLineIndex',
'username_fragment': 'usernameFragment'
}
self._name = name
self._candidate = candidate
self._sdp_mid = sdp_mid
self._sdp_line_index = sdp_line_index
self._username_fragment = username_fragment
@classmethod
def from_dict(cls, dikt) -> 'IceCandidate':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The IceCandidate of this IceCandidate. # noqa: E501
:rtype: IceCandidate
"""
return util.deserialize_model(dikt, cls)
@property
def name(self) -> str:
"""Gets the name of this IceCandidate.
:return: The name of this IceCandidate.
:rtype: str
"""
return self._name
@name.setter
def name(self, name: str):
"""Sets the name of this IceCandidate.
:param name: The name of this IceCandidate.
:type name: str
"""
self._name = name
@property
def candidate(self) -> str:
"""Gets the candidate of this IceCandidate.
:return: The candidate of this IceCandidate.
:rtype: str
"""
return self._candidate
@candidate.setter
def candidate(self, candidate: str):
"""Sets the candidate of this IceCandidate.
:param candidate: The candidate of this IceCandidate.
:type candidate: str
"""
self._candidate = candidate
@property
def sdp_mid(self) -> str:
"""Gets the sdp_mid of this IceCandidate.
:return: The sdp_mid of this IceCandidate.
:rtype: str
"""
return self._sdp_mid
@sdp_mid.setter
def sdp_mid(self, sdp_mid: str):
"""Sets the sdp_mid of this IceCandidate.
:param sdp_mid: The sdp_mid of this IceCandidate.
:type sdp_mid: str
"""
self._sdp_mid = sdp_mid
@property
def sdp_line_index(self) -> int:
"""Gets the sdp_line_index of this IceCandidate.
:return: The sdp_line_index of this IceCandidate.
:rtype: int
"""
return self._sdp_line_index
@sdp_line_index.setter
def sdp_line_index(self, sdp_line_index: int):
"""Sets the sdp_line_index of this IceCandidate.
:param sdp_line_index: The sdp_line_index of this IceCandidate.
:type sdp_line_index: int
"""
self._sdp_line_index = sdp_line_index
@property
def username_fragment(self) -> str:
"""Gets the username_fragment of this IceCandidate.
:return: The username_fragment of this IceCandidate.
:rtype: str
"""
return self._username_fragment
@username_fragment.setter
def username_fragment(self, username_fragment: str):
"""Sets the username_fragment of this IceCandidate.
:param username_fragment: The username_fragment of this IceCandidate.
:type username_fragment: str
"""
self._username_fragment = username_fragment

117
demo_server/openapi_server/models/ice_session_description.py Normal file
View File

@@ -0,0 +1,117 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server import util
class IceSessionDescription(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, name=None, sdp_type=None, sdp=None): # noqa: E501
"""IceSessionDescription - a model defined in OpenAPI
:param name: The name of this IceSessionDescription. # noqa: E501
:type name: str
:param sdp_type: The sdp_type of this IceSessionDescription. # noqa: E501
:type sdp_type: str
:param sdp: The sdp of this IceSessionDescription. # noqa: E501
:type sdp: str
"""
self.openapi_types = {
'name': str,
'sdp_type': str,
'sdp': str
}
self.attribute_map = {
'name': 'name',
'sdp_type': 'sdpType',
'sdp': 'sdp'
}
self._name = name
self._sdp_type = sdp_type
self._sdp = sdp
@classmethod
def from_dict(cls, dikt) -> 'IceSessionDescription':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The IceSessionDescription of this IceSessionDescription. # noqa: E501
:rtype: IceSessionDescription
"""
return util.deserialize_model(dikt, cls)
@property
def name(self) -> str:
"""Gets the name of this IceSessionDescription.
A unique identifier which can be used to send ICE candidates Maps to the session name # noqa: E501
:return: The name of this IceSessionDescription.
:rtype: str
"""
return self._name
@name.setter
def name(self, name: str):
"""Sets the name of this IceSessionDescription.
A unique identifier which can be used to send ICE candidates Maps to the session name # noqa: E501
:param name: The name of this IceSessionDescription.
:type name: str
"""
self._name = name
@property
def sdp_type(self) -> str:
"""Gets the sdp_type of this IceSessionDescription.
Used to construct the remote description in WebRTC # noqa: E501
:return: The sdp_type of this IceSessionDescription.
:rtype: str
"""
return self._sdp_type
@sdp_type.setter
def sdp_type(self, sdp_type: str):
"""Sets the sdp_type of this IceSessionDescription.
Used to construct the remote description in WebRTC # noqa: E501
:param sdp_type: The sdp_type of this IceSessionDescription.
:type sdp_type: str
"""
self._sdp_type = sdp_type
@property
def sdp(self) -> str:
"""Gets the sdp of this IceSessionDescription.
:return: The sdp of this IceSessionDescription.
:rtype: str
"""
return self._sdp
@sdp.setter
def sdp(self, sdp: str):
"""Sets the sdp of this IceSessionDescription.
:param sdp: The sdp of this IceSessionDescription.
:type sdp: str
"""
self._sdp = sdp

117
demo_server/openapi_server/models/knock.py Normal file
View File

@@ -0,0 +1,117 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server.models.ice_session_description import IceSessionDescription
from openapi_server import util
from openapi_server.models.ice_session_description import IceSessionDescription # noqa: E501
class Knock(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, name=None, offer=None, answer=None): # noqa: E501
"""Knock - a model defined in OpenAPI
:param name: The name of this Knock. # noqa: E501
:type name: str
:param offer: The offer of this Knock. # noqa: E501
:type offer: IceSessionDescription
:param answer: The answer of this Knock. # noqa: E501
:type answer: IceSessionDescription
"""
self.openapi_types = {
'name': str,
'offer': IceSessionDescription,
'answer': IceSessionDescription
}
self.attribute_map = {
'name': 'name',
'offer': 'offer',
'answer': 'answer'
}
self._name = name
self._offer = offer
self._answer = answer
@classmethod
def from_dict(cls, dikt) -> 'Knock':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The Knock of this Knock. # noqa: E501
:rtype: Knock
"""
return util.deserialize_model(dikt, cls)
@property
def name(self) -> str:
"""Gets the name of this Knock.
:return: The name of this Knock.
:rtype: str
"""
return self._name
@name.setter
def name(self, name: str):
"""Sets the name of this Knock.
:param name: The name of this Knock.
:type name: str
"""
self._name = name
@property
def offer(self) -> IceSessionDescription:
"""Gets the offer of this Knock.
The service being connected too # noqa: E501
:return: The offer of this Knock.
:rtype: IceSessionDescription
"""
return self._offer
@offer.setter
def offer(self, offer: IceSessionDescription):
"""Sets the offer of this Knock.
The service being connected too # noqa: E501
:param offer: The offer of this Knock.
:type offer: IceSessionDescription
"""
self._offer = offer
@property
def answer(self) -> IceSessionDescription:
"""Gets the answer of this Knock.
:return: The answer of this Knock.
:rtype: IceSessionDescription
"""
return self._answer
@answer.setter
def answer(self, answer: IceSessionDescription):
"""Sets the answer of this Knock.
:param answer: The answer of this Knock.
:type answer: IceSessionDescription
"""
self._answer = answer

63
demo_server/openapi_server/models/list_knocks_response.py Normal file
View File

@@ -0,0 +1,63 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server.models.knock import Knock
from openapi_server import util
from openapi_server.models.knock import Knock # noqa: E501
class ListKnocksResponse(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, knocks=None): # noqa: E501
"""ListKnocksResponse - a model defined in OpenAPI
:param knocks: The knocks of this ListKnocksResponse. # noqa: E501
:type knocks: List[Knock]
"""
self.openapi_types = {
'knocks': List[Knock]
}
self.attribute_map = {
'knocks': 'knocks'
}
self._knocks = knocks
@classmethod
def from_dict(cls, dikt) -> 'ListKnocksResponse':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The ListKnocksResponse of this ListKnocksResponse. # noqa: E501
:rtype: ListKnocksResponse
"""
return util.deserialize_model(dikt, cls)
@property
def knocks(self) -> List[Knock]:
"""Gets the knocks of this ListKnocksResponse.
:return: The knocks of this ListKnocksResponse.
:rtype: List[Knock]
"""
return self._knocks
@knocks.setter
def knocks(self, knocks: List[Knock]):
"""Sets the knocks of this ListKnocksResponse.
:param knocks: The knocks of this ListKnocksResponse.
:type knocks: List[Knock]
"""
self._knocks = knocks

115
demo_server/openapi_server/models/room.py Normal file
View File

@@ -0,0 +1,115 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server.models.server import Server
from openapi_server import util
from openapi_server.models.server import Server # noqa: E501
class Room(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, name=None, display_name=None, servers=None): # noqa: E501
"""Room - a model defined in OpenAPI
:param name: The name of this Room. # noqa: E501
:type name: str
:param display_name: The display_name of this Room. # noqa: E501
:type display_name: str
:param servers: The servers of this Room. # noqa: E501
:type servers: List[Server]
"""
self.openapi_types = {
'name': str,
'display_name': str,
'servers': List[Server]
}
self.attribute_map = {
'name': 'name',
'display_name': 'displayName',
'servers': 'servers'
}
self._name = name
self._display_name = display_name
self._servers = servers
@classmethod
def from_dict(cls, dikt) -> 'Room':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The Room of this Room. # noqa: E501
:rtype: Room
"""
return util.deserialize_model(dikt, cls)
@property
def name(self) -> str:
"""Gets the name of this Room.
:return: The name of this Room.
:rtype: str
"""
return self._name
@name.setter
def name(self, name: str):
"""Sets the name of this Room.
:param name: The name of this Room.
:type name: str
"""
self._name = name
@property
def display_name(self) -> str:
"""Gets the display_name of this Room.
:return: The display_name of this Room.
:rtype: str
"""
return self._display_name
@display_name.setter
def display_name(self, display_name: str):
"""Sets the display_name of this Room.
:param display_name: The display_name of this Room.
:type display_name: str
"""
self._display_name = display_name
@property
def servers(self) -> List[Server]:
"""Gets the servers of this Room.
:return: The servers of this Room.
:rtype: List[Server]
"""
return self._servers
@servers.setter
def servers(self, servers: List[Server]):
"""Sets the servers of this Room.
:param servers: The servers of this Room.
:type servers: List[Server]
"""
self._servers = servers

171
demo_server/openapi_server/models/server.py Normal file
View File

@@ -0,0 +1,171 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server.models.service import Service
from openapi_server import util
from openapi_server.models.service import Service # noqa: E501
class Server(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, name=None, rooms=None, auth_token=None, display_name=None, services=None): # noqa: E501
"""Server - a model defined in OpenAPI
:param name: The name of this Server. # noqa: E501
:type name: str
:param rooms: The rooms of this Server. # noqa: E501
:type rooms: List[str]
:param auth_token: The auth_token of this Server. # noqa: E501
:type auth_token: str
:param display_name: The display_name of this Server. # noqa: E501
:type display_name: str
:param services: The services of this Server. # noqa: E501
:type services: List[Service]
"""
self.openapi_types = {
'name': str,
'rooms': List[str],
'auth_token': str,
'display_name': str,
'services': List[Service]
}
self.attribute_map = {
'name': 'name',
'rooms': 'rooms',
'auth_token': 'authToken',
'display_name': 'displayName',
'services': 'services'
}
self._name = name
self._rooms = rooms
self._auth_token = auth_token
self._display_name = display_name
self._services = services
@classmethod
def from_dict(cls, dikt) -> 'Server':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The Server of this Server. # noqa: E501
:rtype: Server
"""
return util.deserialize_model(dikt, cls)
@property
def name(self) -> str:
"""Gets the name of this Server.
:return: The name of this Server.
:rtype: str
"""
return self._name
@name.setter
def name(self, name: str):
"""Sets the name of this Server.
:param name: The name of this Server.
:type name: str
"""
self._name = name
@property
def rooms(self) -> List[str]:
"""Gets the rooms of this Server.
Tracks which rooms the server should be listed in; this will not be set when a room is listed to preserve privacy of servers. # noqa: E501
:return: The rooms of this Server.
:rtype: List[str]
"""
return self._rooms
@rooms.setter
def rooms(self, rooms: List[str]):
"""Sets the rooms of this Server.
Tracks which rooms the server should be listed in; this will not be set when a room is listed to preserve privacy of servers. # noqa: E501
:param rooms: The rooms of this Server.
:type rooms: List[str]
"""
self._rooms = rooms
@property
def auth_token(self) -> str:
"""Gets the auth_token of this Server.
Used to authenticate requests which access knocks for this server or attempt to update, delete or create services. In future calls, add a header with the format: Authorization: Bearer <auth_token> # noqa: E501
:return: The auth_token of this Server.
:rtype: str
"""
return self._auth_token
@auth_token.setter
def auth_token(self, auth_token: str):
"""Sets the auth_token of this Server.
Used to authenticate requests which access knocks for this server or attempt to update, delete or create services. In future calls, add a header with the format: Authorization: Bearer <auth_token> # noqa: E501
:param auth_token: The auth_token of this Server.
:type auth_token: str
"""
self._auth_token = auth_token
@property
def display_name(self) -> str:
"""Gets the display_name of this Server.
:return: The display_name of this Server.
:rtype: str
"""
return self._display_name
@display_name.setter
def display_name(self, display_name: str):
"""Sets the display_name of this Server.
:param display_name: The display_name of this Server.
:type display_name: str
"""
self._display_name = display_name
@property
def services(self) -> List[Service]:
"""Gets the services of this Server.
:return: The services of this Server.
:rtype: List[Service]
"""
return self._services
@services.setter
def services(self, services: List[Service]):
"""Sets the services of this Server.
:param services: The services of this Server.
:type services: List[Service]
"""
self._services = services

113
demo_server/openapi_server/models/service.py Normal file
View File

@@ -0,0 +1,113 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server import util
class Service(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, name=None, protocol=None, version=None): # noqa: E501
"""Service - a model defined in OpenAPI
:param name: The name of this Service. # noqa: E501
:type name: str
:param protocol: The protocol of this Service. # noqa: E501
:type protocol: str
:param version: The version of this Service. # noqa: E501
:type version: str
"""
self.openapi_types = {
'name': str,
'protocol': str,
'version': str
}
self.attribute_map = {
'name': 'name',
'protocol': 'protocol',
'version': 'version'
}
self._name = name
self._protocol = protocol
self._version = version
@classmethod
def from_dict(cls, dikt) -> 'Service':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The Service of this Service. # noqa: E501
:rtype: Service
"""
return util.deserialize_model(dikt, cls)
@property
def name(self) -> str:
"""Gets the name of this Service.
:return: The name of this Service.
:rtype: str
"""
return self._name
@name.setter
def name(self, name: str):
"""Sets the name of this Service.
:param name: The name of this Service.
:type name: str
"""
self._name = name
@property
def protocol(self) -> str:
"""Gets the protocol of this Service.
:return: The protocol of this Service.
:rtype: str
"""
return self._protocol
@protocol.setter
def protocol(self, protocol: str):
"""Sets the protocol of this Service.
:param protocol: The protocol of this Service.
:type protocol: str
"""
self._protocol = protocol
@property
def version(self) -> str:
"""Gets the version of this Service.
:return: The version of this Service.
:rtype: str
"""
return self._version
@version.setter
def version(self, version: str):
"""Sets the version of this Service.
:param version: The version of this Service.
:type version: str
"""
self._version = version

121
demo_server/openapi_server/models/status.py Normal file
View File

@@ -0,0 +1,121 @@
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from openapi_server.models.base_model import Model
from openapi_server.models.google_protobuf_any import GoogleProtobufAny
from openapi_server import util
from openapi_server.models.google_protobuf_any import GoogleProtobufAny # noqa: E501
class Status(Model):
"""NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
Do not edit the class manually.
"""
def __init__(self, code=None, message=None, details=None): # noqa: E501
"""Status - a model defined in OpenAPI
:param code: The code of this Status. # noqa: E501
:type code: int
:param message: The message of this Status. # noqa: E501
:type message: str
:param details: The details of this Status. # noqa: E501
:type details: List[GoogleProtobufAny]
"""
self.openapi_types = {
'code': int,
'message': str,
'details': List[GoogleProtobufAny]
}
self.attribute_map = {
'code': 'code',
'message': 'message',
'details': 'details'
}
self._code = code
self._message = message
self._details = details
@classmethod
def from_dict(cls, dikt) -> 'Status':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The Status of this Status. # noqa: E501
:rtype: Status
"""
return util.deserialize_model(dikt, cls)
@property
def code(self) -> int:
"""Gets the code of this Status.
The status code, which should be an enum value of [google.rpc.Code][google.rpc.Code]. # noqa: E501
:return: The code of this Status.
:rtype: int
"""
return self._code
@code.setter
def code(self, code: int):
"""Sets the code of this Status.
The status code, which should be an enum value of [google.rpc.Code][google.rpc.Code]. # noqa: E501
:param code: The code of this Status.
:type code: int
"""
self._code = code
@property
def message(self) -> str:
"""Gets the message of this Status.
A developer-facing error message, which should be in English. Any user-facing error message should be localized and sent in the [google.rpc.Status.details][google.rpc.Status.details] field, or localized by the client. # noqa: E501
:return: The message of this Status.
:rtype: str
"""
return self._message
@message.setter
def message(self, message: str):
"""Sets the message of this Status.
A developer-facing error message, which should be in English. Any user-facing error message should be localized and sent in the [google.rpc.Status.details][google.rpc.Status.details] field, or localized by the client. # noqa: E501
:param message: The message of this Status.
:type message: str
"""
self._message = message
@property
def details(self) -> List[GoogleProtobufAny]:
"""Gets the details of this Status.
A list of messages that carry the error details. There is a common set of message types for APIs to use. # noqa: E501
:return: The details of this Status.
:rtype: List[GoogleProtobufAny]
"""
return self._details
@details.setter
def details(self, details: List[GoogleProtobufAny]):
"""Sets the details of this Status.
A list of messages that carry the error details. There is a common set of message types for APIs to use. # noqa: E501
:param details: The details of this Status.
:type details: List[GoogleProtobufAny]
"""
self._details = details

658
demo_server/openapi_server/openapi/openapi.yaml Normal file
View File

@@ -0,0 +1,658 @@
openapi: 3.0.3
info:
title: PeerNetService API
version: 0.0.1
servers:
- url: /
tags:
- name: PeerNetService
paths:
/v1/rooms/{room}:
get:
operationId: peer_net_service_get_room
parameters:
- description: The room id.
explode: false
in: path
name: room
required: true
schema:
type: string
style: simple
responses:
"200":
content:
application/json:
schema:
$ref: '#/components/schemas/Room'
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
/v1/servers:
post:
operationId: peer_net_service_create_server
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/Server'
required: true
responses:
"200":
content:
application/json:
schema:
$ref: '#/components/schemas/Server'
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
/v1/servers/*:
delete:
operationId: peer_net_service_delete_server
parameters:
- explode: true
in: query
name: name
required: false
schema:
type: string
style: form
responses:
"200":
content: {}
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
/v1/servers/{server}/services:
post:
operationId: peer_net_service_create_service
parameters:
- description: The server id.
explode: false
in: path
name: server
required: true
schema:
type: string
style: simple
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/Service'
required: true
responses:
"200":
content:
application/json:
schema:
$ref: '#/components/schemas/Service'
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
/v1/servers/{server}/services/{service}:
delete:
operationId: peer_net_service_delete_service
parameters:
- description: The server id.
explode: false
in: path
name: server
required: true
schema:
type: string
style: simple
- description: The service id.
explode: false
in: path
name: service
required: true
schema:
type: string
style: simple
responses:
"200":
content: {}
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
/v1/servers/{server}/services/{service}/knocks:
get:
operationId: peer_net_service_list_knocks
parameters:
- description: The server id.
explode: false
in: path
name: server
required: true
schema:
type: string
style: simple
- description: The service id.
explode: false
in: path
name: service
required: true
schema:
type: string
style: simple
responses:
"200":
content:
application/json:
schema:
$ref: '#/components/schemas/ListKnocksResponse'
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
post:
description: |-
Creates a knock that will be answered by the peer; the signaler may
delete the knock, regardless of the state, when it ages.
operationId: peer_net_service_create_knock
parameters:
- description: The server id.
explode: false
in: path
name: server
required: true
schema:
type: string
style: simple
- description: The service id.
explode: false
in: path
name: service
required: true
schema:
type: string
style: simple
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/Knock'
required: true
responses:
"200":
content:
application/json:
schema:
$ref: '#/components/schemas/Knock'
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
/v1/servers/{server}/services/{service}/knocks/{knock}:
get:
operationId: peer_net_service_get_knock
parameters:
- description: The server id.
explode: false
in: path
name: server
required: true
schema:
type: string
style: simple
- description: The service id.
explode: false
in: path
name: service
required: true
schema:
type: string
style: simple
- description: The knock id.
explode: false
in: path
name: knock
required: true
schema:
type: string
style: simple
responses:
"200":
content:
application/json:
schema:
$ref: '#/components/schemas/Knock'
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
patch:
operationId: peer_net_service_update_knock
parameters:
- description: The server id.
explode: false
in: path
name: server
required: true
schema:
type: string
style: simple
- description: The service id.
explode: false
in: path
name: service
required: true
schema:
type: string
style: simple
- description: The knock id.
explode: false
in: path
name: knock
required: true
schema:
type: string
style: simple
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/Knock'
required: true
responses:
"200":
content:
application/json:
schema:
$ref: '#/components/schemas/Knock'
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
/v1/sessions/{session}/candidates:
post:
operationId: peer_net_service_create_ice_candidate
parameters:
- description: The session id.
explode: false
in: path
name: session
required: true
schema:
type: string
style: simple
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/IceCandidate'
required: true
responses:
"200":
content:
application/json:
schema:
$ref: '#/components/schemas/IceCandidate'
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
/v1/sessions/{session}/claim/candidates:
get:
description: Acts as both List and Delete atomically.
operationId: peer_net_service_claim_ice_candidates
parameters:
- description: The session id.
explode: false
in: path
name: session
required: true
schema:
type: string
style: simple
responses:
"200":
content:
application/json:
schema:
$ref: '#/components/schemas/ClaimIceCandidatesResponse'
description: OK
default:
content:
application/json:
schema:
$ref: '#/components/schemas/Status'
description: Default error response
tags:
- PeerNetService
x-openapi-router-controller: openapi_server.controllers.peer_net_service_controller
components:
schemas:
ClaimIceCandidatesResponse:
example:
iceCandidates:
- candidate: candidate
usernameFragment: usernameFragment
name: name
sdpMid: sdpMid
sdpLineIndex: 0
- candidate: candidate
usernameFragment: usernameFragment
name: name
sdpMid: sdpMid
sdpLineIndex: 0
properties:
iceCandidates:
items:
$ref: '#/components/schemas/IceCandidate'
title: iceCandidates
type: array
title: ClaimIceCandidatesResponse
type: object
GoogleProtobufAny:
additionalProperties: true
description: Contains an arbitrary serialized message along with a @type that
describes the type of the serialized message.
example:
'@type': '@type'
properties:
'@type':
description: The type of the serialized message.
type: string
title: GoogleProtobufAny
type: object
IceCandidate:
example:
candidate: candidate
usernameFragment: usernameFragment
name: name
sdpMid: sdpMid
sdpLineIndex: 0
properties:
name:
title: name
type: string
candidate:
title: candidate
type: string
sdpMid:
title: sdpMid
type: string
sdpLineIndex:
format: int32
title: sdpLineIndex
type: integer
usernameFragment:
title: usernameFragment
type: string
title: IceCandidate
type: object
IceSessionDescription:
example:
name: name
sdpType: sdpType
sdp: sdp
properties:
name:
description: |-
A unique identifier which can be used to send ICE candidates
Maps to the session name
title: name
type: string
sdpType:
description: Used to construct the remote description in WebRTC
title: sdpType
type: string
sdp:
title: sdp
type: string
title: IceSessionDescription
type: object
Knock:
example:
offer: ""
answer:
name: name
sdpType: sdpType
sdp: sdp
name: name
properties:
name:
title: name
type: string
offer:
allOf:
- $ref: '#/components/schemas/IceSessionDescription'
description: The service being connected too
title: offer
answer:
$ref: '#/components/schemas/IceSessionDescription'
title: Knock
type: object
ListKnocksResponse:
example:
knocks:
- offer: ""
answer:
name: name
sdpType: sdpType
sdp: sdp
name: name
- offer: ""
answer:
name: name
sdpType: sdpType
sdp: sdp
name: name
properties:
knocks:
items:
$ref: '#/components/schemas/Knock'
title: knocks
type: array
title: ListKnocksResponse
type: object
Room:
example:
servers:
- rooms:
- rooms
- rooms
displayName: displayName
authToken: authToken
name: name
services:
- protocol: protocol
name: name
version: version
- protocol: protocol
name: name
version: version
- rooms:
- rooms
- rooms
displayName: displayName
authToken: authToken
name: name
services:
- protocol: protocol
name: name
version: version
- protocol: protocol
name: name
version: version
displayName: displayName
name: name
properties:
name:
title: name
type: string
displayName:
title: displayName
type: string
servers:
items:
$ref: '#/components/schemas/Server'
title: servers
type: array
title: Room
type: object
Server:
example:
rooms:
- rooms
- rooms
displayName: displayName
authToken: authToken
name: name
services:
- protocol: protocol
name: name
version: version
- protocol: protocol
name: name
version: version
properties:
name:
title: name
type: string
rooms:
description: |-
Tracks which rooms the server should be listed in; this will not
be set when a room is listed to preserve privacy of servers.
items:
type: string
title: rooms
type: array
writeOnly: true
authToken:
description: |-
Used to authenticate requests which access knocks for this server
or attempt to update, delete or create services.
In future calls, add a header with the format:
Authorization: Bearer <auth_token>
title: authToken
type: string
writeOnly: true
displayName:
title: displayName
type: string
services:
items:
$ref: '#/components/schemas/Service'
title: services
type: array
title: Server
type: object
Service:
example:
protocol: protocol
name: name
version: version
properties:
name:
title: name
type: string
protocol:
title: protocol
type: string
version:
title: version
type: string
title: Service
type: object
Status:
description: "The `Status` type defines a logical error model that is suitable\
\ for different programming environments, including REST APIs and RPC APIs.\
\ It is used by [gRPC](https://github.com/grpc). Each `Status` message contains\
\ three pieces of data: error code, error message, and error details. You\
\ can find out more about this error model and how to work with it in the\
\ [API Design Guide](https://cloud.google.com/apis/design/errors)."
example:
code: 0
details:
- '@type': '@type'
- '@type': '@type'
message: message
properties:
code:
description: "The status code, which should be an enum value of [google.rpc.Code][google.rpc.Code]."
format: int32
title: code
type: integer
message:
description: "A developer-facing error message, which should be in English.\
\ Any user-facing error message should be localized and sent in the [google.rpc.Status.details][google.rpc.Status.details]\
\ field, or localized by the client."
title: message
type: string
details:
description: A list of messages that carry the error details. There is
a common set of message types for APIs to use.
items:
$ref: '#/components/schemas/GoogleProtobufAny'
title: details
type: array
title: Status
type: object

16
demo_server/openapi_server/test/__init__.py Normal file
View File

@@ -0,0 +1,16 @@
import logging
import connexion
from flask_testing import TestCase
from openapi_server.encoder import JSONEncoder
class BaseTestCase(TestCase):
def create_app(self):
logging.getLogger('connexion.operation').setLevel('ERROR')
app = connexion.App(__name__, specification_dir='../openapi/')
app.app.json_encoder = JSONEncoder
app.add_api('openapi.yaml', pythonic_params=True)
return app.app

200
demo_server/openapi_server/test/test_peer_net_service_controller.py Normal file
View File

@@ -0,0 +1,200 @@
import unittest
from flask import json
from openapi_server.models.claim_ice_candidates_response import ClaimIceCandidatesResponse # noqa: E501
from openapi_server.models.ice_candidate import IceCandidate # noqa: E501
from openapi_server.models.knock import Knock # noqa: E501
from openapi_server.models.list_knocks_response import ListKnocksResponse # noqa: E501
from openapi_server.models.room import Room # noqa: E501
from openapi_server.models.server import Server # noqa: E501
from openapi_server.models.service import Service # noqa: E501
from openapi_server.models.status import Status # noqa: E501
from openapi_server.test import BaseTestCase
class TestPeerNetServiceController(BaseTestCase):
"""PeerNetServiceController integration test stubs"""
def test_peer_net_service_claim_ice_candidates(self):
"""Test case for peer_net_service_claim_ice_candidates
"""
query_string = [('name', 'name_example')]
headers = {
'Accept': 'application/json',
}
response = self.client.open(
'/v1/sessions/{session}/claim/candidates'.format(session='session_example'),
method='GET',
headers=headers,
query_string=query_string)
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_peer_net_service_create_ice_candidate(self):
"""Test case for peer_net_service_create_ice_candidate
"""
ice_candidate = {"candidate":"candidate","usernameFragment":"usernameFragment","sdpMid":"sdpMid","sdpLineIndex":0}
query_string = [('targetSession', 'target_session_example')]
headers = {
'Accept': 'application/json',
'Content-Type': 'application/json',
}
response = self.client.open(
'/v1/sessions/{session}/candidates'.format(session='session_example'),
method='POST',
headers=headers,
data=json.dumps(ice_candidate),
content_type='application/json',
query_string=query_string)
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_peer_net_service_create_knock(self):
"""Test case for peer_net_service_create_knock
"""
knock = {"offer":{"name":"name","sdpType":"sdpType","sdp":"sdp"},"answer":{"name":"name","sdpType":"sdpType","sdp":"sdp"},"service":"","name":"name"}
headers = {
'Accept': 'application/json',
'Content-Type': 'application/json',
}
response = self.client.open(
'/v1/servers/{server}/sevices/{sevice}/knocks'.format(server='server_example', sevice='sevice_example'),
method='POST',
headers=headers,
data=json.dumps(knock),
content_type='application/json')
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_peer_net_service_create_server(self):
"""Test case for peer_net_service_create_server
"""
server = {"rooms":["rooms","rooms"],"displayName":"displayName","authToken":"authToken","name":"name","services":[{"protocol":"protocol","name":"name","version":"version"},{"protocol":"protocol","name":"name","version":"version"}]}
headers = {
'Accept': 'application/json',
'Content-Type': 'application/json',
}
response = self.client.open(
'/v1/servers',
method='POST',
headers=headers,
data=json.dumps(server),
content_type='application/json')
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_peer_net_service_create_service(self):
"""Test case for peer_net_service_create_service
"""
return
service = {"protocol":"protocol","name":"name","version":"version"}
headers = {
'Accept': 'application/json',
'Content-Type': 'application/json',
}
response = self.client.open(
'/v1/servers/{server}/services'.format(server='server_example'),
method='POST',
headers=headers,
data=json.dumps(service),
content_type='application/json')
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_peer_net_service_delete_server(self):
"""Test case for peer_net_service_delete_server
"""
query_string = [('name', 'name_example')]
headers = {
'Accept': 'application/json',
}
response = self.client.open(
'/v1/servers/*',
method='DELETE',
headers=headers,
query_string=query_string)
self.assert404(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_peer_net_service_delete_service(self):
"""Test case for peer_net_service_delete_service
"""
return
headers = {
'Accept': 'application/json',
}
response = self.client.open(
'/v1/servers/{server}/services/{service}'.format(server='server_example', service='service_example'),
method='DELETE',
headers=headers)
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_peer_net_service_get_room(self):
"""Test case for peer_net_service_get_room
"""
return
headers = {
'Accept': 'application/json',
}
response = self.client.open(
'/v1/rooms/{room}'.format(room='room_example'),
method='GET',
headers=headers)
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_peer_net_service_list_knocks(self):
"""Test case for peer_net_service_list_knocks
"""
headers = {
'Accept': 'application/json',
}
response = self.client.open(
'/v1/servers/{server}/services/{service}/knocks'.format(server='server_example', service='service_example'),
method='GET',
headers=headers)
self.assert401(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_peer_net_service_update_knock(self):
"""Test case for peer_net_service_update_knock
"""
knock2 = {"offer":{"name":"name","sdpType":"sdpType","sdp":"sdp"},"answer":{"name":"name","sdpType":"sdpType","sdp":"sdp"},"service":"","name":"name"}
headers = {
'Accept': 'application/json',
'Content-Type': 'application/json',
}
response = self.client.open(
'/v1/servers/{server}/services/{service}/knocks/{knock}'.format(server='server_example', service='service_example', knock='knock_example'),
method='PATCH',
headers=headers,
data=json.dumps(knock2),
content_type='application/json')
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
if __name__ == '__main__':
unittest.main()

30
demo_server/openapi_server/typing_utils.py Normal file
View File

@@ -0,0 +1,30 @@
import sys
if sys.version_info < (3, 7):
import typing
def is_generic(klass):
""" Determine whether klass is a generic class """
return type(klass) == typing.GenericMeta
def is_dict(klass):
""" Determine whether klass is a Dict """
return klass.__extra__ == dict
def is_list(klass):
""" Determine whether klass is a List """
return klass.__extra__ == list
else:
def is_generic(klass):
""" Determine whether klass is a generic class """
return hasattr(klass, '__origin__')
def is_dict(klass):
""" Determine whether klass is a Dict """
return klass.__origin__ == dict
def is_list(klass):
""" Determine whether klass is a List """
return klass.__origin__ == list

147
demo_server/openapi_server/util.py Normal file
View File

@@ -0,0 +1,147 @@
import datetime
import typing
from openapi_server import typing_utils
def _deserialize(data, klass):
"""Deserializes dict, list, str into an object.
:param data: dict, list or str.
:param klass: class literal, or string of class name.
:return: object.
"""
if data is None:
return None
if klass in (int, float, str, bool, bytearray):
return _deserialize_primitive(data, klass)
elif klass == object:
return _deserialize_object(data)
elif klass == datetime.date:
return deserialize_date(data)
elif klass == datetime.datetime:
return deserialize_datetime(data)
elif typing_utils.is_generic(klass):
if typing_utils.is_list(klass):
return _deserialize_list(data, klass.__args__[0])
if typing_utils.is_dict(klass):
return _deserialize_dict(data, klass.__args__[1])
else:
return deserialize_model(data, klass)
def _deserialize_primitive(data, klass):
"""Deserializes to primitive type.
:param data: data to deserialize.
:param klass: class literal.
:return: int, long, float, str, bool.
:rtype: int | long | float | str | bool
"""
try:
value = klass(data)
except UnicodeEncodeError:
value = data
except TypeError:
value = data
return value
def _deserialize_object(value):
"""Return an original value.
:return: object.
"""
return value
def deserialize_date(string):
"""Deserializes string to date.
:param string: str.
:type string: str
:return: date.
:rtype: date
"""
if string is None:
return None
try:
from dateutil.parser import parse
return parse(string).date()
except ImportError:
return string
def deserialize_datetime(string):
"""Deserializes string to datetime.
The string should be in iso8601 datetime format.
:param string: str.
:type string: str
:return: datetime.
:rtype: datetime
"""
if string is None:
return None
try:
from dateutil.parser import parse
return parse(string)
except ImportError:
return string
def deserialize_model(data, klass):
"""Deserializes list or dict to model.
:param data: dict, list.
:type data: dict | list
:param klass: class literal.
:return: model object.
"""
instance = klass()
if not instance.openapi_types:
return data
for attr, attr_type in instance.openapi_types.items():
if data is not None \
and instance.attribute_map[attr] in data \
and isinstance(data, (list, dict)):
value = data[instance.attribute_map[attr]]
setattr(instance, attr, _deserialize(value, attr_type))
return instance
def _deserialize_list(data, boxed_type):
"""Deserializes a list and its elements.
:param data: list to deserialize.
:type data: list
:param boxed_type: class literal.
:return: deserialized list.
:rtype: list
"""
return [_deserialize(sub_data, boxed_type)
for sub_data in data]
def _deserialize_dict(data, boxed_type):
"""Deserializes a dict and its elements.
:param data: dict to deserialize.
:type data: dict
:param boxed_type: class literal.
:return: deserialized dict.
:rtype: dict
"""
return {k: _deserialize(v, boxed_type)
for k, v in data.items() }