mart-w/kanidm
1
0
Fork 0
mirror of https://github.com/kanidm/kanidm.git synced 2025-02-23 12:37:00 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

Improvements to integration test

Browse source
This commit is contained in:
William Brown 2018-11-26 16:13:22 +10:00
parent 2e554833db
commit 66bf9bcc3c
11 changed files with 312 additions and 225 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.toml
View file

@ -20,6 +20,7 @@ actix = "0.7"
actix-web = "0.7"
bytes = "0.4"
env_logger = "0.5"
reqwest = "0.9"
tokio = "0.1"

2
src/be/mod.rs
View file

@ -154,7 +154,7 @@ impl Backend {
}
// Take filter, and AuditEvent ref?
pub fn search(&self, au: &mut AuditEvent, filt: &Filter) -> Result<Vec<Entry>, ()> {
pub fn search(&self, au: &mut AuditEvent, filt: &Filter) -> Result<Vec<Entry>, BackendError> {
// Do things
// Alloc a vec for the entries.
// FIXME: Make this actually a good size for the result set ...

21
src/config.rs
View file

@ -0,0 +1,21 @@
#[derive(Serialize, Deserialize, Debug)]
pub struct Configuration {
pub address: String,
pub threads: usize,
pub db_path: String,
pub maximum_request: usize,
// db type later
}
impl Configuration {
pub fn new() -> Self {
Configuration {
address: String::from("127.0.0.1:8080"),
threads: 8,
db_path: String::from(""),
maximum_request: 262144, // 256k
// log type
// log path
}
}
}

175
src/core.rs Normal file
View file

@ -0,0 +1,175 @@
use actix::SystemRunner;
use actix_web::{
error, http, middleware, App, AsyncResponder, Error, FutureResponse, HttpMessage, HttpRequest,
HttpResponse, Path, State,
};
use bytes::BytesMut;
use futures::{future, Future, Stream};
use super::config::Configuration;
use super::event::{CreateEvent, EventResult, SearchEvent};
use super::filter::Filter;
use super::log;
use super::proto::{CreateRequest, SearchRequest};
use super::server;
struct AppState {
qe: actix::Addr<server::QueryServer>,
max_size: usize,
}
macro_rules! json_event_decode {
($req:expr, $state:expr, $event_type:ty, $message_type:ty) => {{
// This is copied every request. Is there a better way?
// The issue is the fold move takes ownership of state if
// we don't copy this here
let max_size = $state.max_size;
// HttpRequest::payload() is stream of Bytes objects
$req.payload()
// `Future::from_err` acts like `?` in that it coerces the error type from
// the future into the final error type
.from_err()
// `fold` will asynchronously read each chunk of the request body and
// call supplied closure, then it resolves to result of closure
.fold(BytesMut::new(), move |mut body, chunk| {
// limit max size of in-memory payload
if (body.len() + chunk.len()) > max_size {
Err(error::ErrorBadRequest("overflow"))
} else {
body.extend_from_slice(&chunk);
Ok(body)
}
})
// `Future::and_then` can be used to merge an asynchronous workflow with a
// synchronous workflow
.and_then(
move |body| -> Box<Future<Item = HttpResponse, Error = Error>> {
// body is loaded, now we can deserialize serde-json
// let r_obj = serde_json::from_slice::<SearchRequest>(&body);
let r_obj = serde_json::from_slice::<$message_type>(&body);
// Send to the db for create
match r_obj {
Ok(obj) => {
let res = $state
.qe
.send(
// Could make this a .into_inner() and move?
// event::SearchEvent::new(obj.filter),
<($event_type)>::new(obj),
)
.from_err()
.and_then(|res| match res {
Ok(entries) => Ok(HttpResponse::Ok().json(entries)),
Err(e) => Ok(HttpResponse::InternalServerError().json(e)),
});
Box::new(res)
}
Err(e) => Box::new(future::err(error::ErrorBadRequest(format!(
"Json Decode Failed: {:?}",
e
)))),
}
},
)
}};
}
// Handle the various end points we need to expose
/// simple handle
fn index(req: &HttpRequest<AppState>) -> HttpResponse {
println!("{:?}", req);
HttpResponse::Ok().body("Hello\n")
}
fn class_list((_name, state): (Path<String>, State<AppState>)) -> FutureResponse<HttpResponse> {
// println!("request to class_list");
let filt = Filter::Pres(String::from("objectclass"));
state
.qe
.send(
// This is where we need to parse the request into an event
// LONG TERM
// Make a search REQUEST, and create the audit struct here, then
// pass it to the server
//
// FIXME: Don't use SEARCHEVENT here!!!!
//
SearchEvent::new(SearchRequest::new(filt)),
)
// TODO: How to time this part of the code?
// What does this do?
.from_err()
.and_then(|res| match res {
// What type is entry?
Ok(EventResult::Search { entries }) => Ok(HttpResponse::Ok().json(entries)),
Ok(_) => Ok(HttpResponse::Ok().into()),
// Can we properly report this?
Err(_) => Ok(HttpResponse::InternalServerError().into()),
})
// What does this do?
.responder()
}
fn create(
(req, state): (HttpRequest<AppState>, State<AppState>),
) -> impl Future<Item = HttpResponse, Error = Error> {
json_event_decode!(req, state, CreateEvent, CreateRequest)
}
fn search(
(req, state): (HttpRequest<AppState>, State<AppState>),
) -> impl Future<Item = HttpResponse, Error = Error> {
json_event_decode!(req, state, SearchEvent, SearchRequest)
}
pub fn create_server_core(config: Configuration) {
// Configure the middleware logger
::std::env::set_var("RUST_LOG", "actix_web=info");
env_logger::init();
// Until this point, we probably want to write to stderr
// Start up the logging system: for now it just maps to stderr
let log_addr = log::start();
log_event!(log_addr, "Starting rsidm with configuration: {:?}", config);
// Start the query server with the given be path: future config
let server_addr = server::start(log_addr.clone(), config.db_path.as_str(), config.threads);
// Copy the max size
let max_size = config.maximum_request;
// start the web server
actix_web::server::new(move || {
App::with_state(AppState {
qe: server_addr.clone(),
max_size: max_size,
})
// Connect all our end points here.
.middleware(middleware::Logger::default())
.resource("/", |r| r.f(index))
// curl --header "Content-Type: application/json" --request POST --data '{ "entries": [ {"attrs": {"class": ["group"], "name": ["testgroup"], "description": ["testperson"]}}]}' http://127.0.0.1:8080/create
.resource("/create", |r| {
r.method(http::Method::POST).with_async(create)
})
// curl --header "Content-Type: application/json" --request POST --data '{ "filter" : { "Eq": ["class", "user"] }}' http://127.0.0.1:8080/search
.resource("/search", |r| {
r.method(http::Method::POST).with_async(search)
})
// Add an ldap compat search function type?
.resource("/list/{class_list}", |r| {
r.method(http::Method::GET).with(class_list)
})
.resource("/list/{class_list}/", |r| {
r.method(http::Method::GET).with(class_list)
})
})
.bind(config.address)
.unwrap()
.start();
}

7
src/error.rs
View file

@ -8,3 +8,10 @@ pub enum SchemaError {
InvalidAttributeSyntax,
EmptyFilter,
}
#[derive(Serialize, Deserialize, Debug, PartialEq)]
pub enum OperationError {
EmptyRequest,
Backend,
SchemaViolation,
}

5
src/event.rs
View file

@ -2,6 +2,7 @@ use super::filter::Filter;
use super::proto::{CreateRequest, SearchRequest};
use actix::prelude::*;
use entry::Entry;
use error::OperationError;
// Should the event Result have the log items?
// FIXME: Remove seralising here - each type should
@ -24,7 +25,7 @@ pub struct SearchEvent {
}
impl Message for SearchEvent {
type Result = Result<EventResult, ()>;
type Result = Result<EventResult, OperationError>;
}
impl SearchEvent {
@ -47,7 +48,7 @@ pub struct CreateEvent {
}
impl Message for CreateEvent {
type Result = Result<EventResult, ()>;
type Result = Result<EventResult, OperationError>;
}
impl CreateEvent {

5
src/lib.rs
View file

@ -12,6 +12,9 @@ extern crate r2d2_sqlite;
extern crate rusqlite;
extern crate uuid;
extern crate bytes;
extern crate env_logger;
// use actix::prelude::*;
// use actix_web::{
// http, middleware, App, AsyncResponder, FutureResponse, HttpRequest, HttpResponse, Path, State,
@ -25,6 +28,8 @@ pub mod log;
#[macro_use]
mod audit;
mod be;
pub mod config;
pub mod core;
pub mod entry;
pub mod error;
pub mod event;

184
src/main.rs
View file

@ -1,187 +1,19 @@
extern crate actix;
extern crate actix_web;
extern crate bytes;
extern crate env_logger;
extern crate futures;
extern crate serde;
extern crate serde_derive;
extern crate serde_json;
extern crate uuid;
// use actix::prelude::*;
use actix_web::{
error, http, middleware, App, AsyncResponder, Error, FutureResponse, HttpMessage, HttpRequest,
HttpResponse, Path, State,
};
use bytes::BytesMut;
use futures::{future, Future, Stream};
#[macro_use]
extern crate actix;
extern crate rsidm;
use rsidm::event::{CreateEvent, EventResult, SearchEvent};
use rsidm::filter::Filter;
use rsidm::log;
use rsidm::proto::{CreateRequest, SearchRequest};
use rsidm::server;
const MAX_SIZE: usize = 262_144; //256k - this is the upper bound on create/search etc.
struct AppState {
qe: actix::Addr<server::QueryServer>,
}
macro_rules! json_event_decode {
($req:expr, $state:expr, $event_type:ty, $message_type:ty) => {{
// HttpRequest::payload() is stream of Bytes objects
$req.payload()
// `Future::from_err` acts like `?` in that it coerces the error type from
// the future into the final error type
.from_err()
// `fold` will asynchronously read each chunk of the request body and
// call supplied closure, then it resolves to result of closure
.fold(BytesMut::new(), move |mut body, chunk| {
// limit max size of in-memory payload
if (body.len() + chunk.len()) > MAX_SIZE {
Err(error::ErrorBadRequest("overflow"))
} else {
body.extend_from_slice(&chunk);
Ok(body)
}
})
// `Future::and_then` can be used to merge an asynchronous workflow with a
// synchronous workflow
.and_then(
move |body| -> Box<Future<Item = HttpResponse, Error = Error>> {
// body is loaded, now we can deserialize serde-json
// let r_obj = serde_json::from_slice::<SearchRequest>(&body);
let r_obj = serde_json::from_slice::<$message_type>(&body);
// Send to the db for create
match r_obj {
Ok(obj) => {
let res = $state
.qe
.send(
// Could make this a .into_inner() and move?
// event::SearchEvent::new(obj.filter),
<($event_type)>::new(obj),
)
.from_err()
.and_then(|res| match res {
Ok(entries) => Ok(HttpResponse::Ok().json(entries)),
Err(_) => Ok(HttpResponse::InternalServerError().into()),
});
Box::new(res)
}
Err(e) => Box::new(future::err(error::ErrorBadRequest(format!(
"Json Decode Failed: {:?}",
e
)))),
}
},
)
}};
}
// Handle the various end points we need to expose
/// simple handle
fn index(req: &HttpRequest<AppState>) -> HttpResponse {
println!("{:?}", req);
HttpResponse::Ok().body("Hello\n")
}
fn class_list((_name, state): (Path<String>, State<AppState>)) -> FutureResponse<HttpResponse> {
// println!("request to class_list");
let filt = Filter::Pres(String::from("objectclass"));
state
.qe
.send(
// This is where we need to parse the request into an event
// LONG TERM
// Make a search REQUEST, and create the audit struct here, then
// pass it to the server
//
// FIXME: Don't use SEARCHEVENT here!!!!
//
SearchEvent::new(SearchRequest::new(filt)),
)
// TODO: How to time this part of the code?
// What does this do?
.from_err()
.and_then(|res| match res {
// What type is entry?
Ok(EventResult::Search { entries }) => Ok(HttpResponse::Ok().json(entries)),
Ok(_) => Ok(HttpResponse::Ok().into()),
// Can we properly report this?
Err(_) => Ok(HttpResponse::InternalServerError().into()),
})
// What does this do?
.responder()
}
fn create(
(req, state): (HttpRequest<AppState>, State<AppState>),
) -> impl Future<Item = HttpResponse, Error = Error> {
json_event_decode!(req, state, CreateEvent, CreateRequest)
}
fn search(
(req, state): (HttpRequest<AppState>, State<AppState>),
) -> impl Future<Item = HttpResponse, Error = Error> {
json_event_decode!(req, state, SearchEvent, SearchRequest)
}
use rsidm::config::Configuration;
use rsidm::core::create_server_core;
fn main() {
// Configure the middleware logger
::std::env::set_var("RUST_LOG", "actix_web=info");
env_logger::init();
let sys = actix::System::new("rsidm-server");
// read the config (if any?)
// How do we make the config accesible to all threads/workers? clone it?
// Make it an Arc<Config>?
// Until this point, we probably want to write to stderr
// Start up the logging system: for now it just maps to stderr
let log_addr = log::start();
// FIXME: Pass config to the server core
let config = Configuration::new();
let sys = actix::System::new("rsidm-server");
// Start the query server with the given be path: future config
let server_addr = server::start(log_addr.clone(), "test.db", 8);
// start the web server
actix_web::server::new(move || {
App::with_state(AppState {
qe: server_addr.clone(),
})
// Connect all our end points here.
.middleware(middleware::Logger::default())
.resource("/", |r| r.f(index))
// curl --header "Content-Type: application/json" --request POST --data '{ "entries": [ {"attrs": {"class": ["group"], "name": ["testgroup"], "description": ["testperson"]}}]}' http://127.0.0.1:8080/create
.resource("/create", |r| {
r.method(http::Method::POST).with_async(create)
})
// curl --header "Content-Type: application/json" --request POST --data '{ "filter" : { "Eq": ["class", "user"] }}' http://127.0.0.1:8080/search
.resource("/search", |r| {
r.method(http::Method::POST).with_async(search)
})
// Add an ldap compat search function type?
.resource("/list/{class_list}", |r| {
r.method(http::Method::GET).with(class_list)
})
.resource("/list/{class_list}/", |r| {
r.method(http::Method::GET).with(class_list)
})
})
.bind("127.0.0.1:8080")
.unwrap()
.start();
log_event!(log_addr, "Starting rsidm on http://127.0.0.1:8080");
create_server_core(config);
let _ = sys.run();
}

5
src/proto.rs
View file

@ -3,6 +3,11 @@ use super::filter::Filter;
// These proto implementations are here because they have public definitions
// FIXME: We probably need a proto entry to transform our
// server core entry into.
// FIXME: Proto Response as well here
#[derive(Debug, Serialize, Deserialize)]
pub struct SearchRequest {
pub filter: Filter,

47
src/server.rs
View file

@ -1,8 +1,10 @@
use actix::prelude::*;
use audit::AuditEvent;
use be::Backend;
use be::{Backend, BackendError};
use entry::Entry;
use error::OperationError;
use event::{CreateEvent, EventResult, SearchEvent};
use log::EventLog;
use schema::Schema;
@ -56,24 +58,33 @@ impl QueryServer {
// Actually conduct a search request
// This is the core of the server, as it processes the entire event
// applies all parts required in order and more.
pub fn search(&mut self, au: &mut AuditEvent, se: &SearchEvent) -> Result<Vec<Entry>, ()> {
match self.be.search(au, &se.filter) {
Ok(r) => Ok(r),
Err(_) => Err(()),
}
pub fn search(
&mut self,
au: &mut AuditEvent,
se: &SearchEvent,
) -> Result<Vec<Entry>, OperationError> {
let res = self
.be
.search(au, &se.filter)
.map(|r| r)
.map_err(|_| OperationError::Backend);
// We'll add ACI later
res
}
// What should this take?
// This should probably take raw encoded entries? Or sohuld they
// be handled by fe?
pub fn create(&mut self, au: &mut AuditEvent, ce: &CreateEvent) -> Result<(), ()> {
pub fn create(&mut self, au: &mut AuditEvent, ce: &CreateEvent) -> Result<(), OperationError> {
// Start a txn
// Run any pre checks
// FIXME: Normalise all entries incoming
let r = ce.entries.iter().fold(Ok(()), |acc, e| {
if acc.is_ok() {
self.schema.validate_entry(e).map_err(|_| ())
self.schema
.validate_entry(e)
.map_err(|_| OperationError::SchemaViolation)
} else {
acc
}
@ -83,12 +94,18 @@ impl QueryServer {
}
// We may change from ce.entries later to something else?
match self.be.create(au, &ce.entries) {
Ok(_) => Ok(()),
Err(_) => Err(()),
}
let res = self
.be
.create(au, &ce.entries)
.map(|_| ())
.map_err(|e| match e {
BackendError::EmptyRequest => OperationError::EmptyRequest,
_ => OperationError::Backend,
});
// Run and post checks
// Commit/Abort the txn
res
}
}
@ -108,7 +125,7 @@ impl Actor for QueryServer {
// at this point our just is just to route to do_<action>
impl Handler<SearchEvent> for QueryServer {
type Result = Result<EventResult, ()>;
type Result = Result<EventResult, OperationError>;
fn handle(&mut self, msg: SearchEvent, _: &mut Self::Context) -> Self::Result {
let mut audit = AuditEvent::new();
@ -135,7 +152,7 @@ impl Handler<SearchEvent> for QueryServer {
}
impl Handler<CreateEvent> for QueryServer {
type Result = Result<EventResult, ()>;
type Result = Result<EventResult, OperationError>;
fn handle(&mut self, msg: CreateEvent, _: &mut Self::Context) -> Self::Result {
let mut audit = AuditEvent::new();
@ -147,7 +164,7 @@ impl Handler<CreateEvent> for QueryServer {
Err(e) => Err(e),
};
audit_log!(audit, "End create event {:?}", msg);
audit_log!(audit, "End create event {:?} -> {:?}", msg, res);
audit.end_event("create");
// At the end of the event we send it for logging.
self.log.do_send(audit);

77
tests/integration_test.rs
View file

@ -2,14 +2,24 @@ extern crate actix;
use actix::prelude::*;
extern crate rsidm;
use rsidm::config::Configuration;
use rsidm::core::create_server_core;
use rsidm::entry::Entry;
use rsidm::event::EventResult;
use rsidm::log::{self, EventLog, LogEvent};
use rsidm::proto::{CreateRequest, SearchRequest};
use rsidm::server::{self, QueryServer};
// use be;
extern crate reqwest;
extern crate futures;
use futures::future;
use futures::future::Future;
use std::sync::mpsc;
use std::thread;
extern crate tokio;
// use tokio::executor::current_thread::CurrentThread;
@ -17,48 +27,61 @@ extern crate tokio;
macro_rules! run_test {
($test_fn:expr) => {{
System::run(|| {
let (tx, rx) = mpsc::channel();
thread::spawn(|| {
// setup
// Create a server config in memory for use - use test settings
// Create a log: In memory - for now it's just stdout
let test_log = log::start();
// Create the db as a temporary, see:
// https://sqlite.org/inmemorydb.html
let test_server = server::start(test_log.clone(), "", 1);
System::run(move || {
let config = Configuration::new();
create_server_core(config);
// This appears to be bind random ...
// let srv = srv.bind("127.0.0.1:0").unwrap();
let _ = tx.send(System::current());
});
});
let sys = rx.recv().unwrap();
System::set_current(sys.clone());
// Do we need any fixtures?
// Yes probably, but they'll need to be futures as well ...
// later we could accept fixture as it's own future for re-use
// For now these can also bypass the FE code
// let fixture_fut = ();
$test_fn();
// We have to spawn every test as a future
let fut = $test_fn(test_log, test_server);
// Now chain them ...
// Now append the server shutdown.
let comp_fut = fut.map_err(|_| ()).and_then(|_r| {
println!("Stopping actix ...");
actix::System::current().stop();
future::result(Ok(()))
});
// Run the future
tokio::spawn(comp_fut);
// We DO NOT need teardown, as sqlite is in mem
// let the tables hit the floor
});
let _ = sys.stop();
}};
}
#[test]
fn test_schema() {
run_test!(
|log: actix::Addr<EventLog>, _server: actix::Addr<QueryServer>| log.send(LogEvent {
msg: String::from("Test log event")
})
);
fn test_server_proto() {
run_test!(|| {
let client = reqwest::Client::new();
let c = CreateRequest {
entries: Vec::new(),
};
let mut response = client
.post("http://127.0.0.1:8080/create")
.body(serde_json::to_string(&c).unwrap())
.send()
.unwrap();
println!("{:?}", response);
let r: EventResult = serde_json::from_str(response.text().unwrap().as_str()).unwrap();
println!("{:?}", r);
// deserialise the response here
// check it's valid.
()
});
}
/*