text: /* -*- C -*- */
  /*
   * Copyright (c) 2017-2020 Seagate Technology LLC and/or its Affiliates
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   *
   * For any questions about this software or licensing,
   * please email opensource@seagate.com or cortx-questions@seagate.com.
   *
   */
  
  
  
  #define M0_TRACE_SUBSYSTEM M0_TRACE_SUBSYS_FDMI
  #include "lib/trace.h"
  
  #include "ut/ut.h"
  #include "fdmi/module.h"
  #include "fdmi/fdmi.h"
  #include "fdmi/fops.h"
  #include "fdmi/service.h"
  #include "fdmi/plugin_dock_internal.h"
  #include "fdmi/ut/sd_common.h"
  #include "lib/string.h"
  #include "lib/finject.h"
  #include "rpc/item_internal.h"                /* m0_rpc_item_change_state */
  #include "rpc/packet_internal.h"              /* m0_rpc_packet_traverse_items */
  #include "rpc/rpc_internal.h"                 /* item2conn */
  #include "rpc/rpc_machine_internal.h"
  #include "rpc/conn_pool_internal.h"
  
  /*----------------------------------------
    fdmi_pd_register_filter
    ----------------------------------------*/
  
  static
  int pd_ut_pcb_fdmi_rec(struct m0_uint128   *rec_id,
  		       struct m0_buf        fdmi_rec,
  		       struct m0_fid        filter_id);
  
  void fdmi_pd_register_filter(void)
  {
  	const struct m0_fid              ffid = {
  		.f_container = 0xDEA110C,
  		.f_key       = 0xDA221ED,
  	};
  	const struct m0_fdmi_plugin_ops  pcb = {
  		.po_fdmi_rec = pd_ut_pcb_fdmi_rec
  	};
  	const struct m0_fdmi_filter_desc fd;
  	const struct m0_fdmi_pd_ops     *pdo = m0_fdmi_plugin_dock_api_get();
  	struct m0_fdmi_filter_reg       *freg;
  	struct m0_fid                    fids[2] = { [0] = ffid, [1] = { 0, 0 } };
  	int                              rc;
  
  	rc = (pdo->fpo_register_filter)(&ffid, &fd, NULL);
  	M0_UT_ASSERT(rc == -EINVAL);
  
  	m0_fi_enable_off_n_on_m("m0_alloc", "fail_allocation", 0, 1);
  	rc = (pdo->fpo_register_filter)(&ffid, &fd, &pcb);
  	M0_UT_ASSERT(rc == -ENOMEM);
  
  	m0_fi_enable_off_n_on_m("m0_alloc", "fail_allocation", 1, 1);
  	rc = (pdo->fpo_register_filter)(&ffid, &fd, &pcb);
  	M0_UT_ASSERT(rc == -ENOMEM);
  
  	m0_fi_enable_off_n_on_m("m0_alloc", "fail_allocation", 2, 1);
  	rc = (pdo->fpo_register_filter)(&ffid, &fd, &pcb);
  	M0_UT_ASSERT(rc == -ENOMEM);
  
  	m0_fi_disable("m0_alloc", "fail_allocation");
  
  	(pdo->fpo_enable_filters)(true, fids, ARRAY_SIZE(fids));
  
  	/* add filter */
  	rc = (pdo->fpo_register_filter)(&ffid, &fd, &pcb);
  	M0_UT_ASSERT(rc == 0);
  
  	/* find filter and see if it's disabled */
  	freg = m0_fdmi__pdock_filter_reg_find(&ffid);
  	M0_UT_ASSERT(freg != NULL);
  	M0_UT_ASSERT(freg->ffr_flags & M0_BITS(M0_FDMI_FILTER_INACTIVE));
  
  	/* toggle active flag back and forth */
  	if (freg->ffr_flags & M0_BITS(M0_FDMI_FILTER_INACTIVE)) {
  
  		(pdo->fpo_enable_filters)(true, fids, ARRAY_SIZE(fids));
  		M0_UT_ASSERT(!(freg->ffr_flags &
  			       M0_BITS(M0_FDMI_FILTER_INACTIVE)));
  
  		(pdo->fpo_enable_filters)(false, fids, ARRAY_SIZE(fids));
  		M0_UT_ASSERT(!!(freg->ffr_flags &
  				M0_BITS(M0_FDMI_FILTER_INACTIVE)));
  	}
  
  	/* deregister filter */
  	(pdo->fpo_deregister_plugin)(fids, ARRAY_SIZE(fids));
  }
  
  /*----------------------------------------
    fdmi_pd_fom_norpc
    ----------------------------------------*/
  
  static struct m0_semaphore     g_sem;
  struct m0_uint128              frid_new = M0_UINT128(0xEEEE, 0xEEEE);
  struct m0_uint128             *frid_watch;
  
  
  static void ut_pd_fom_fini(struct m0_fom *fom);
  
  int (*native_create)(struct m0_fop  *fop,
  		     struct m0_fom **out,
  		     struct m0_reqh *reqh) = NULL;
  
  int detour_create(struct m0_fop *fop, struct m0_fom **out, struct m0_reqh *reqh)
  {
  	int                        rc;
  	struct pdock_fom          *pd_fom;
  	struct m0_fom             *fom;
  	struct m0_fdmi_record_reg *rreg;
  	struct m0_fop_fdmi_record *frec;
  
  	M0_ENTRY();
  
  	m0_fi_enable_off_n_on_m("m0_alloc", "fail_allocation", 0, 1);
  	rc = (*native_create)(fop, out, reqh);
  	M0_UT_ASSERT(rc == -ENOMEM);
  
  	m0_fi_enable_off_n_on_m("m0_alloc", "fail_allocation", 1, 1);
  	rc = (*native_create)(fop, out, reqh);
  	M0_UT_ASSERT(rc == -ENOMEM);
  
  	m0_fi_enable_off_n_on_m("m0_alloc", "fail_allocation", 2, 1);
  	rc = (*native_create)(fop, out, reqh);
  	M0_UT_ASSERT(rc == -ENOENT);
  
  	m0_fi_enable_off_n_on_m("m0_alloc", "fail_allocation", 3, 1);
  	rc = (*native_create)(fop, out, reqh);
  	M0_UT_ASSERT(rc == -ENOMEM);
  
  	m0_fi_disable("m0_alloc", "fail_allocation");
  
  	rc = (*native_create)(fop, out, reqh);
  	M0_UT_ASSERT(m0_fop_data(fop)
  		     ? rc == 0
  		     : rc == -ENOMEM);
  
  	if (rc != 0) {
  		m0_semaphore_up(&g_sem);
  		return rc;
  	}
  
  	M0_UT_ASSERT(out != NULL);
  
  	fom = *out;
  	M0_UT_ASSERT(fom != NULL);
  
  	pd_fom = container_of(fom, struct pdock_fom, pf_fom);
  	pd_fom->pf_custom_fom_fini = ut_pd_fom_fini;
  
  	frec = m0_fop_data(fop);
  	M0_UT_ASSERT(frec != NULL);
  	rreg = m0_fdmi__pdock_record_reg_find(&frec->fr_rec_id);
  	M0_UT_ASSERT(rreg != NULL);
  
  	return M0_RC(rc);
  }
  
  struct m0_fom_type_ops    fomt_ops = { .fto_create = detour_create };
  
  
  #define FID_CTN   0xDEA110C
  #define FID_KEY   0xDA221ED
  #define FID_CTN1  0x1FA11ED
  #define FID_KEY1  0x1A110CA
  #define M0_FID(c_, k_)  { .f_container = c_, .f_key = k_ }
  
  struct m0_fid     ffid = M0_FID(FID_CTN, FID_KEY);
  struct m0_uint128 frid = M0_UINT128(0xBEC02, 0xF11ED);
  
  static void ut_pd_fom_fini(struct m0_fom *fom)
  {
  	struct m0_fop_fdmi_record *frec;
  	struct m0_fdmi_record_reg *rreg;
  
  	M0_ENTRY();
  
  	M0_PRE(m0_fom_phase(fom) == M0_FOM_PHASE_FINISH);
  	m0_fi_disable("m0_fdmi__pdock_record_reg_find", "fail_find");
  
  	frec = m0_fop_data(fom->fo_fop);
  	M0_UT_ASSERT(m0_uint128_eq(frid_watch, &frec->fr_rec_id));
  
  	rreg = m0_fdmi__pdock_record_reg_find(&frec->fr_rec_id);
  	if (rreg != NULL)
  		m0_ref_put(&rreg->frr_ref);
  
  	/* detach fop from fom */
  	m0_free0(&fom->fo_fop);
  
  	/* kill fom */
  	m0_fom_fini(fom);
  
  	m0_semaphore_up(&g_sem);
  
  
  	M0_LEAVE();
  }
  
  struct m0_fid             ffids[2] = {
  	[0] = M0_FID(FID_CTN, FID_KEY),
  	[1] = M0_FID(FID_CTN1, FID_KEY1)
  };
  struct m0_fdmi_flt_id_arr farr     = {
  	.fmf_flt_id = ffids,
  	.fmf_count = ARRAY_SIZE(ffids)
  };
  
  static int pd_ut_pcb_fdmi_rec(struct m0_uint128   *rec_id,
  			      struct m0_buf        fdmi_rec,
  			      struct m0_fid        filter_id)
  {
  	struct m0_fdmi_record_reg *rreg;
  
  	M0_ENTRY("fdmi record arrived: id " U128X_F
  		 ", filter id = " FID_SF,
  		 U128_P(rec_id),
  		 FID_P(&filter_id));
  
  	M0_UT_ASSERT(m0_uint128_eq(frid_watch, rec_id));
  	M0_UT_ASSERT(m0_fid_eq(&ffids[0], &filter_id) ||
  		     m0_fid_eq(&ffids[1], &filter_id));
  
  	rreg = m0_fdmi__pdock_record_reg_find(rec_id);
  	M0_UT_ASSERT(rreg != NULL);
  
  	if (m0_uint128_eq(&frid_new, &rreg->frr_rec->fr_rec_id)) {
  		m0_fi_enable_off_n_on_m("m0_fdmi__pdock_record_reg_find",
  					"fail_find", 0, 1);
  	}
  
  	return 0;
  }
  
  void __fdmi_pd_fom_norpc(bool register_filter)
  {
  	struct m0_reqh                  *reqh;
  	struct m0_fop                   *fop;
  	struct m0_fop_fdmi_record       *rec;
  	struct m0_fop_type               fopt;
  	int                              rc;
  
  	struct m0_fdmi_filter_reg       *freg;
  	const struct m0_fdmi_plugin_ops  pcb = {
  		.po_fdmi_rec = pd_ut_pcb_fdmi_rec
  	};
  	const struct m0_fdmi_filter_desc fd;
  	const struct m0_fdmi_pd_ops     *pdo = m0_fdmi_plugin_dock_api_get();
  	struct m0_fid                    all_fids[3] = {
  						[0] = ffid,
  						[1] = ffids[1],
  						[2] = { 0, 0 } };
  
  	M0_ENTRY();
  
  	if (register_filter) {
  		/* add filter */
  		rc = (pdo->fpo_register_filter)(&ffid, &fd, &pcb);
  		M0_UT_ASSERT(rc == 0);
  		freg = m0_fdmi__pdock_filter_reg_find(&ffid);
  		M0_UT_ASSERT(freg != NULL);
  		M0_UT_ASSERT(freg->ffr_flags &
  			     M0_BITS(M0_FDMI_FILTER_INACTIVE));
  		(pdo->fpo_enable_filters)(true, ffids, ARRAY_SIZE(ffids));
  		M0_UT_ASSERT(!(freg->ffr_flags &
  			       M0_BITS(M0_FDMI_FILTER_INACTIVE)));
  		rc = (pdo->fpo_register_filter)(&ffids[1], &fd, &pcb);
  		M0_UT_ASSERT(rc == 0);
  		freg = m0_fdmi__pdock_filter_reg_find(&ffid);
  		M0_UT_ASSERT(freg != NULL);
  		/*
  		 * Intentionally free it, otherwise memleak, because we
  		 * are manually set it to NULL.
  		 */
  		m0_free(freg->ffr_pcb);
  		freg->ffr_pcb = NULL;
  	}
  
  	/* intercept fom create */
  	fopt = m0_fop_fdmi_rec_not_fopt;
  	native_create = fopt.ft_fom_type.ft_ops->fto_create;
  	fopt.ft_fom_type = (struct m0_fom_type) {
  /* hook it up*/
  		.ft_ops        =
  			&fomt_ops,
  /* keep native*/
  		.ft_conf       =
  			m0_fop_fdmi_rec_not_fopt.ft_fom_type.ft_conf,
  		.ft_state_conf =
  			m0_fop_fdmi_rec_not_fopt.ft_fom_type.ft_state_conf,
  		.ft_rstype     =
  			m0_fop_fdmi_rec_not_fopt.ft_fom_type.ft_rstype,
  	};
  
  	fdmi_serv_start_ut(&filterc_stub_ops);
  
  	m0_semaphore_init(&g_sem, 0);
  
  	reqh = &g_sd_ut.motr.cc_reqh_ctx.rc_reqh;
  
  	/* try with known fdmi record id */
  	frid_watch = &frid;
  	M0_ALLOC_PTR(rec);
  	M0_UT_ASSERT(rec != NULL);
  	M0_SET0(rec);
  	rec->fr_rec_id       = *frid_watch;
  	rec->fr_rec_type     = M0_FDMI_REC_TYPE_FOL;
  	rec->fr_matched_flts = farr;
  
  	fop = m0_fop_alloc(&fopt, rec, (void*)1);
  	M0_UT_ASSERT(fop != NULL);
  	fop->f_item.ri_rmachine = NULL;
  
  	rc = m0_reqh_fop_handle(reqh, fop);
  	M0_UT_ASSERT(rc == 0);
  
  	/* wait for fom finishing */
  	m0_semaphore_down(&g_sem);
  	m0_free(rec);
  
  	/* try with another fdmi record id */
  	frid_watch = &frid_new;
  	M0_ALLOC_PTR(rec);
  	M0_UT_ASSERT(rec != NULL);
  	M0_SET0(rec);
  	rec->fr_rec_id       = *frid_watch;
  	rec->fr_rec_type     = M0_FDMI_REC_TYPE_FOL;
  	rec->fr_matched_flts = farr;
  
  	fop = m0_fop_alloc(&fopt, rec, (void*)1);
  	M0_UT_ASSERT(fop != NULL);
  	fop->f_item.ri_rmachine = NULL;
  
  	rc = m0_reqh_fop_handle(reqh, fop);
  	M0_UT_ASSERT(rc == 0);
  
  	/* wait for fom finishing */
  	m0_semaphore_down(&g_sem);
  	m0_free(rec);
  
  	fdmi_serv_stop_ut();
  
  	m0_semaphore_fini(&g_sem);
  	if (register_filter) {
  		/* deregister filter */
  		(pdo->fpo_deregister_plugin)(all_fids, ARRAY_SIZE(all_fids));
  	}
  	M0_LEAVE();
  }
  
  void fdmi_pd_fom_norpc(void)
  {
  	__fdmi_pd_fom_norpc(false);
  	__fdmi_pd_fom_norpc(true);
  }
  
  /*----------------------------------------
    fdmi_pd_rec_inject_fini
    ----------------------------------------*/
  
  void fdmi_pd_rec_inject_fini(void) {
  	struct m0_fop              *fop;
  	struct m0_fdmi_record_reg  *rreg;
  	struct m0_fop_fdmi_record  *rec;
  
  	M0_ALLOC_PTR(rec);
  	M0_UT_ASSERT(rec != NULL);
  	M0_SET0(rec);
  	rec->fr_rec_id       = frid;
  	rec->fr_rec_type     = M0_FDMI_REC_TYPE_FOL;
  	rec->fr_matched_flts = farr;
  
  	fop = m0_fop_alloc(&m0_fop_fdmi_rec_not_fopt, rec, (void*)1);
  	M0_UT_ASSERT(fop != NULL);
  	fop->f_item.ri_rmachine = NULL;
  
  	rreg = m0_fdmi__pdock_fdmi_record_register(fop);
  	M0_UT_ASSERT(rreg != NULL);
  
  	rreg->frr_ep_addr = m0_strdup("test");
  
  	m0_fdmi__plugin_dock_fini();
  	m0_free(rec);
  	m0_free(fop);
  
  	/* Init pd back, its used in other tests. */
  	m0_fdmi__plugin_dock_init();
  }
  
  /*----------------------------------------
    fdmi_pd_fake_rec_reg
    ----------------------------------------*/
  
  static
  struct pdock_client_conn {
  	struct m0_rpc_conn    pc_conn;
  	struct m0_rpc_session pc_sess;
  }  g_cc;
  static struct test_rpc_env         g_rpc_env;
  static struct m0_rpc_packet       *g_rpc_packet;
  
  static void my_item_done(struct m0_rpc_packet *p,
  			 struct m0_rpc_item *item, int rc)
  {
  	m0_rpc_item_change_state(item, M0_RPC_ITEM_SENT);
  	m0_rpc_item_timer_stop(item);
  	m0_rpc_conn_ha_timer_stop(item2conn(item));
  }
  
  static int my_packet_ready(struct m0_rpc_packet *p)
  {
  	g_rpc_packet = p;
  	m0_rpc_packet_traverse_items(p, &my_item_done, 0);
  	return 0;
  }
  
  
  const struct m0_rpc_frm_ops frm_ops = {
  	.fo_packet_ready = my_packet_ready,
  };
  
  
  extern struct fdmi_sd_ut_ctx       g_sd_ut;
  extern const struct m0_filterc_ops filterc_send_notif_ops;
  
  
  void fdmi_pd_fake_rec_reg(void)
  {
  	struct m0_fop             *fop;
  
  	M0_ENTRY();
  
  	fdmi_serv_start_ut(&filterc_send_notif_ops);
  	prepare_rpc_env(&g_rpc_env, &g_sd_ut.motr.cc_reqh_ctx.rc_reqh,
  			&frm_ops, true, &g_cc.pc_conn, &g_cc.pc_sess);
  	fop = m0_fop_alloc(&m0_fop_fdmi_rec_not_fopt, NULL,
  			   &g_rpc_env.tre_rpc_machine);
  	M0_UT_ASSERT(fop != 0);
  	m0_fop_to_rpc_item(fop)->ri_session  = &g_cc.pc_sess;
  	m0_free(fop);
  	unprepare_rpc_env(&g_rpc_env);
  	fdmi_serv_stop_ut();
  
  	M0_LEAVE();
  }
  
  /*----------------------------------------
    fdmi_pd_fake_rec_release
    ----------------------------------------*/
  
  void fdmi_pd_fake_rec_release(void)
  {
  	struct m0_fop               *fop;
  	struct m0_fdmi_record_reg   *rreg;
  	struct m0_fop_fdmi_record   *rec;
  	const struct m0_fdmi_pd_ops *pdo = m0_fdmi_plugin_dock_api_get();
  	struct m0_uint128            zero_frid = M0_UINT128(0, 0);
  
  	M0_ENTRY();
  
  	M0_ALLOC_PTR(rec);
  	M0_UT_ASSERT(rec != NULL);
  	M0_SET0(rec);
  	rec->fr_rec_id       = frid;
  	rec->fr_rec_type     = M0_FDMI_REC_TYPE_FOL;
  	rec->fr_matched_flts = farr;
  
  	fdmi_serv_start_ut(&filterc_send_notif_ops);
  
  	prepare_rpc_env(&g_rpc_env, &g_sd_ut.motr.cc_reqh_ctx.rc_reqh,
  			&frm_ops, true, &g_cc.pc_conn, &g_cc.pc_sess);
  
  	fop = m0_fop_alloc(&m0_fop_fdmi_rec_not_fopt, rec,
  			   &g_rpc_env.tre_rpc_machine);
  	M0_UT_ASSERT(fop != 0);
  
  	m0_fop_to_rpc_item(fop)->ri_session  = &g_cc.pc_sess;
  
  	m0_fi_enable_off_n_on_m("m0_fdmi__pdock_fdmi_record_register",
  				"fail_fdmi_rec_reg", 0, 1);
  	rreg = m0_fdmi__pdock_fdmi_record_register(fop);
  	M0_UT_ASSERT(rreg == NULL);
  	m0_fi_disable("m0_fdmi__pdock_fdmi_record_register",
  		      "fail_fdmi_rec_reg");
  
  	rreg = m0_fdmi__pdock_fdmi_record_register(fop);
  	M0_UT_ASSERT(rreg != NULL);
  
  	/* lock reg to not allow release to happen */
  	m0_ref_get(&rreg->frr_ref);
  	M0_UT_ASSERT(m0_ref_read(&rreg->frr_ref) == 2);
  
  	/* unknown record id */
  	(*pdo->fpo_release_fdmi_rec)(&zero_frid, &ffid);
  	M0_UT_ASSERT(m0_ref_read(&rreg->frr_ref) == 2);
  
  	/* known record id */
  	(*pdo->fpo_release_fdmi_rec)(&frid, &ffid);
  	M0_UT_ASSERT(m0_ref_read(&rreg->frr_ref) == 1);
  
  	m0_free(fop);
  	m0_free(rec);
  
  	unprepare_rpc_env(&g_rpc_env);
  	fdmi_serv_stop_ut();
  
  	M0_LEAVE();
  }
  
  /*----------------------------------------
    fdmi_pd_fake_release_rep
    ----------------------------------------*/
  
  extern struct m0_rpc_item_ops release_ri_ops;
  extern struct m0_fop_type m0_fop_fdmi_rec_release_rep_fopt;
  
  void fdmi_pd_fake_release_rep(void)
  {
  	struct m0_fop              *fop;
  	struct m0_fop              *rep_fop;
  	struct m0_fdmi_record_reg  *rreg;
  	struct m0_fop_fdmi_record  *rec;
  	struct m0_rpc_item         *item;
  
  	struct m0_fop_fdmi_rec_release rr;
  
  	M0_ENTRY();
  
  	M0_ALLOC_PTR(rec);
  	M0_UT_ASSERT(rec != NULL);
  	M0_SET0(rec);
  	rec->fr_rec_id       = frid;
  	rec->fr_rec_type     = M0_FDMI_REC_TYPE_FOL;
  	rec->fr_matched_flts = farr;
  
  	fdmi_serv_start_ut(&filterc_send_notif_ops);
  
  	prepare_rpc_env(&g_rpc_env, &g_sd_ut.motr.cc_reqh_ctx.rc_reqh,
  			&frm_ops, true, &g_cc.pc_conn, &g_cc.pc_sess);
  
  	/* imitate fdmi rec notification fop */
  	fop = m0_fop_alloc(&m0_fop_fdmi_rec_not_fopt, rec,
  			   &g_rpc_env.tre_rpc_machine);
  	M0_UT_ASSERT(fop != 0);
  
  	item = m0_fop_to_rpc_item(fop);
  	item->ri_session  = &g_cc.pc_sess;
  
  	rreg = m0_fdmi__pdock_fdmi_record_register(fop);
  	M0_UT_ASSERT(rreg != NULL);
  
  	rreg->frr_fop = fop;
  	rreg->frr_sess = item->ri_session;
  	rreg->frr_ep_addr = m0_strdup(m0_rpc_item_remote_ep_addr(item));
  
  	/* imitate reply fop */
  	rr.frr_frid = frid;
  
  	rep_fop = m0_fop_alloc(&m0_fop_fdmi_rec_release_rep_fopt, &rr,
  			       &g_rpc_env.tre_rpc_machine);
  	M0_UT_ASSERT(fop != 0);
  
  	item = m0_fop_to_rpc_item(rep_fop);
  	item->ri_session  = &g_cc.pc_sess;
  
  	m0_sm_group_lock(&item->ri_rmachine->rm_sm_grp);
  
  	/* imitate successful rpc reply */
  	rreg->frr_sess = NULL;
   	(release_ri_ops.rio_replied)(m0_fop_to_rpc_item(rep_fop));
  
  	/* imitate failed rpc reply */
  	m0_fop_to_rpc_item(rep_fop)->ri_error = -1;
   	(release_ri_ops.rio_replied)(m0_fop_to_rpc_item(rep_fop));
  
  	m0_sm_group_unlock(&item->ri_rmachine->rm_sm_grp);
  
  	m0_free(rep_fop);
  
  	unprepare_rpc_env(&g_rpc_env);
  	fdmi_serv_stop_ut();
  
  	M0_LEAVE();
  }
  
  /*----------------------------------------
    fdmi_pd_fake_release_nomem
   ----------------------------------------*/
  struct m0_rpc_conn_pool *ut_pdock_conn_pool(void);
  
  void fdmi_pd_fake_release_nomem()
  {
  	struct m0_fop               *fop;
  	struct m0_fdmi_record_reg   *rreg;
  	struct m0_fdmi_record_reg   *rreg_test;
  	struct m0_fop_fdmi_record   *rec;
  	const struct m0_fdmi_pd_ops *pdo = m0_fdmi_plugin_dock_api_get();
  	struct m0_rpc_conn_pool     *conn_pool = ut_pdock_conn_pool();
  	struct m0_rpc_conn_pool_item *pool_item;
  
  	m0_fi_disable("m0_alloc", "fail_allocation");
  
  	M0_ALLOC_PTR(rec);
  	M0_UT_ASSERT(rec != NULL);
  	M0_SET0(rec);
  	rec->fr_rec_id       = M0_UINT128(0x1111, 0x2222);
  	rec->fr_rec_type     = M0_FDMI_REC_TYPE_FOL;
  	rec->fr_matched_flts = farr;
  
  
  	fdmi_serv_start_ut(&filterc_send_notif_ops);
  
  	/* injecting fake conn */
  	M0_ALLOC_PTR(pool_item);
  	rpc_conn_pool_items_tlink_init_at_tail(pool_item,
  					       &conn_pool->cp_items);
  
  	fop = m0_fop_alloc(&m0_fop_fdmi_rec_not_fopt, rec,
  			   &g_rpc_env.tre_rpc_machine);
  	M0_UT_ASSERT(fop != 0);
  
  	m0_fop_to_rpc_item(fop)->ri_session = &pool_item->cpi_rpc_link.rlk_sess;
  	m0_fop_to_rpc_item(fop)->ri_rmachine = &g_rpc_env.tre_rpc_machine;
  
  	prepare_rpc_env(&g_rpc_env, &g_sd_ut.motr.cc_reqh_ctx.rc_reqh,
  			&frm_ops, true,
  			&pool_item->cpi_rpc_link.rlk_conn,
  			&pool_item->cpi_rpc_link.rlk_sess);
  
  	rreg = m0_fdmi__pdock_fdmi_record_register(fop);
  	M0_UT_ASSERT(rreg != NULL);
  
  	/* lock reg to not allow release to happen */
  	M0_UT_ASSERT(m0_ref_read(&rreg->frr_ref) == 1);
  
  	m0_fi_enable_off_n_on_m("m0_alloc", "fail_allocation", 1, 1);
  	(*pdo->fpo_release_fdmi_rec)(&rec->fr_rec_id, &ffid);
  
  	/* record remains registered */
  	rreg_test = m0_fdmi__pdock_record_reg_find(&rec->fr_rec_id);
  	M0_UT_ASSERT(rreg_test != NULL);
  	M0_UT_ASSERT(m0_ref_read(&rreg_test->frr_ref) == 0);
  
  	m0_fi_disable("m0_alloc", "fail_allocation");
  
  
  	m0_fi_enable_off_n_on_m("m0_rpc_conn_pool_get", "fail_conn_get", 0, 1);
  	m0_ref_get(&rreg->frr_ref);
  	(*pdo->fpo_release_fdmi_rec)(&rec->fr_rec_id, &ffid);
  
  	/* record remains registered */
  	rreg_test = m0_fdmi__pdock_record_reg_find(&rec->fr_rec_id);
  	M0_UT_ASSERT(rreg_test != NULL);
  	M0_UT_ASSERT(m0_ref_read(&rreg_test->frr_ref) == 0);
  
  	m0_fi_disable("m0_rpc_conn_pool_get", "fail_conn_get");
  
  
  	m0_fi_enable_off_n_on_m("m0_alloc", "fail_allocation", 0, 1);
  	m0_ref_get(&rreg->frr_ref);
  	(*pdo->fpo_release_fdmi_rec)(&rec->fr_rec_id, &ffid);
  
  	/* record remains registered */
  	rreg_test = m0_fdmi__pdock_record_reg_find(&rec->fr_rec_id);
  	M0_UT_ASSERT(rreg_test != NULL);
  	M0_UT_ASSERT(m0_ref_read(&rreg_test->frr_ref) == 0);
  
  	m0_fi_disable("m0_alloc", "fail_allocation");
  
  
  	m0_ref_get(&rreg->frr_ref);
  
  	m0_free(fop);
  
  	unprepare_rpc_env(&g_rpc_env);
  
  	rpc_conn_pool_items_tlink_del_fini(pool_item);
  	m0_free(pool_item);
  	pool_item = NULL;
  
  	fdmi_serv_stop_ut();
  }
  
  struct m0_ut_suite fdmi_pd_ut = {
  	.ts_name = "fdmi-pd-ut",
  	.ts_tests = {
  		{ "fdmi-pd-register-filter",    fdmi_pd_register_filter    },
  		{ "fdmi-pd-fom-norpc",          fdmi_pd_fom_norpc          },
  		{ "fdmi-pd-rec-inject-fini",    fdmi_pd_rec_inject_fini    },
  		{ "fdmi-pd-fake-release-nomem", fdmi_pd_fake_release_nomem },
  		{ "fdmi-pd-fake-release-rep",   fdmi_pd_fake_release_rep   },
  		{ "fdmi-pd-fake-rec-release",   fdmi_pd_fake_rec_release   },
  		{ "fdmi-pd-fake-rec-reg",       fdmi_pd_fake_rec_reg       },
  		{ NULL, NULL },
  	},
  };
  
  #undef M0_TRACE_SUBSYSTEM
  
  /*
   *  Local variables:
   *  c-indentation-style: "K&R"
   *  c-basic-offset: 8
   *  tab-width: 8
   *  fill-column: 80
   *  scroll-step: 1
   *  End:
   */
  /*
   * vim: tabstop=8 shiftwidth=8 noexpandtab textwidth=80 nowrap
   */
  